From 64243bfb396277c0764e3f8e5129778c78b910c1 Mon Sep 17 00:00:00 2001
From: anlicheng <244108715@qq.com>
Date: Thu, 7 May 2026 23:21:21 +0800
Subject: [PATCH] =?UTF-8?q?=E8=B0=83=E6=95=B4=E5=92=8Ciot=E7=9A=84?=
 =?UTF-8?q?=E9=80=9A=E8=AE=AF=E6=9C=BA=E5=88=B6?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 docs/container_command_docker_json.md | 737 ++++++++++++++++++++++++++
 docs/efka_iot_protocol.md             | 255 +++++++++
 src/transport/efka_client.erl         |  98 ++--
 3 files changed, 1052 insertions(+), 38 deletions(-)
 create mode 100644 docs/container_command_docker_json.md
 create mode 100644 docs/efka_iot_protocol.md

diff --git a/docs/container_command_docker_json.md b/docs/container_command_docker_json.md
new file mode 100644
index 0000000..bf3708a
--- /dev/null
+++ b/docs/container_command_docker_json.md
@@ -0,0 +1,737 @@
+# IOT 容器命令到 Docker JSON 的转换说明
+
+本文档描述 `efka` 收到 `iot` 下发的容器管理 command 后，接受的 Erlang map 格式，以及 deploy 命令中 `create` 参数如何转换成 Docker Engine API 接收的 JSON。
+
+对应代码：
+
+- command 接收入口：[src/transport/efka_client.erl](/usr/local/code/cloudkit/efka/src/transport/efka_client.erl:177)
+- 部署任务管理：[src/docker/docker_deploy_manager.erl](/usr/local/code/cloudkit/efka/src/docker/docker_deploy_manager.erl:36)
+- 部署执行：[src/docker/docker_deployer.erl](/usr/local/code/cloudkit/efka/src/docker/docker_deployer.erl:39)
+- Docker JSON 构造：[src/docker/docker_container_builder.erl](/usr/local/code/cloudkit/efka/src/docker/docker_container_builder.erl:14)
+- Docker API 调用：[src/docker/docker_commands.erl](/usr/local/code/cloudkit/efka/src/docker/docker_commands.erl:36)
+
+## 1. 协议入口
+
+`iot` 通过 TLS 长连接向 `efka` 下发容器命令：
+
+```erlang
+{command, Ref, {container, CommandMap}}
+```
+
+`efka` 执行后回复：
+
+```erlang
+{command_response, Ref, {container, Reply}}
+```
+
+`Reply` 取值：
+
+```erlang
+ok
+{ok, Result}
+{error, Reason}
+```
+
+只有 `efka_client` 处于 `activated` 状态时，容器命令才会正常执行；处于 `restricted` 或其他状态时会返回错误。
+
+## 2. 容器命令 map
+
+### list
+
+```erlang
+#{action => list}
+```
+
+执行：
+
+```erlang
+docker_commands:get_containers()
+```
+
+Docker API：
+
+```http
+GET /containers/json?all=true
+```
+
+### deploy
+
+```erlang
+#{
+    action => deploy,
+    task_id => TaskId,
+    params => Params
+}
+```
+
+执行：
+
+```erlang
+docker_deploy_manager:deploy(TaskId, Params)
+```
+
+`deploy/2` 会启动独立部署进程，HTTP command response 只表示部署任务是否成功启动。实际部署过程和结果通过 `task_event` 消息流上报给 `iot`。
+
+### start
+
+```erlang
+#{
+    action => start,
+    target => Target
+}
+```
+
+执行：
+
+```erlang
+docker_commands:start_container(ContainerNameOrId)
+```
+
+Docker API：
+
+```http
+POST /containers/{name_or_id}/start
+```
+
+### stop
+
+```erlang
+#{
+    action => stop,
+    target => Target,
+    timeout_seconds => TimeoutSeconds
+}
+```
+
+执行：
+
+```erlang
+docker_commands:stop_container(ContainerNameOrId, TimeoutSeconds)
+```
+
+Docker API：
+
+```http
+POST /containers/{name_or_id}/stop?t={TimeoutSeconds}
+```
+
+### kill
+
+```erlang
+#{
+    action => kill,
+    target => Target,
+    signal => Signal
+}
+```
+
+执行：
+
+```erlang
+docker_commands:kill_container(ContainerNameOrId, Signal)
+```
+
+### remove
+
+```erlang
+#{
+    action => remove,
+    target => Target,
+    force => Force,
+    remove_volumes => RemoveVolumes
+}
+```
+
+执行：
+
+```erlang
+docker_commands:remove_container(ContainerNameOrId, Force, RemoveVolumes)
+```
+
+### config
+
+```erlang
+#{
+    action => config,
+    target => Target,
+    config => Config
+}
+```
+
+执行：
+
+```erlang
+docker_helper:update_container_config(ContainerNameOrId, iolist_to_binary(Config))
+```
+
+该命令不会调用 Docker API，只会更新 `efka` 主机上对应容器目录里的 `service.conf`。
+
+## 3. Target 解析规则
+
+`target` 是 map：
+
+```erlang
+#{
+    name => ContainerName,
+    id => ContainerId
+}
+```
+
+解析规则：
+
+- 优先使用 `name`。
+- `name` 为空时使用 `id`。
+- `name` 和 `id` 都为空时会触发匹配错误。
+
+## 4. deploy params 格式
+
+`deploy` 的 `params` 必须包含：
+
+```erlang
+#{
+    container_name => ContainerName,
+    container_dir => ContainerDir,
+    create => Create
+}
+```
+
+字段说明：
+
+| 字段 | 类型 | 说明 |
+| --- | --- | --- |
+| `container_name` | binary | 容器名称。用于容器创建 URL 的 `name` 参数，也会注入环境变量 `CONTAINER_NAME`。 |
+| `container_dir` | binary | 容器在 efka 主机上的应用目录。为空时使用 `efka.root_dir/ContainerName/`。 |
+| `create` | map | Docker create options 的中间结构，由 `iot` 侧构造，`efka` 侧补丁后转为 Docker JSON。 |
+
+`docker_deploy_manager` 会根据 `container_name` 和 `container_dir` 确保容器目录存在，并写入 `.container_dir` 指针文件。
+
+## 5. create 中间结构
+
+`create` 的结构：
+
+```erlang
+#{
+    config => ContainerConfig,
+    host_config => HostConfig,
+    networking_config => NetworkingConfig
+}
+```
+
+这三个 map 会被 `docker_container_builder:build_options/3` 转成 Docker Engine API JSON。
+
+## 6. efka 自动补丁规则
+
+在转 Docker JSON 前，`efka` 会先执行补丁：
+
+### 环境变量
+
+向 `create.config.env` 前置注入：
+
+```erlang
+<<"CONTAINER_NAME=", ContainerName/binary>>
+```
+
+如果原 env 列表已经包含完全相同的值，则不会重复添加。
+
+### 配置文件 volume
+
+向 `create.config.volumes` 前置注入容器内配置路径：
+
+```erlang
+<<"/usr/local/etc/service.conf">>
+```
+
+### 配置文件 bind
+
+向 `create.host_config.binds` 前置注入宿主机配置文件映射：
+
+```erlang
+<<ConfigFile/binary, ":/usr/local/etc/service.conf">>
+```
+
+其中 `ConfigFile` 是当前容器目录下的：
+
+```text
+service.conf
+```
+
+如果列表里已存在完全相同的 bind，则不会重复添加。
+
+## 7. Docker JSON 顶层结构
+
+`docker_commands:create_container/2` 最终调用：
+
+```http
+POST /containers/create?name={container_name}
+Content-Type: application/json
+```
+
+请求 body 来自：
+
+```erlang
+Options = docker_container_builder:build_options(ContainerName, ContainerDir, Create),
+Body = iolist_to_binary(json:encode(Options))
+```
+
+最终 JSON 顶层字段：
+
+```json
+{
+  "Image": "...",
+  "Cmd": [],
+  "Entrypoint": [],
+  "Env": [],
+  "Labels": {},
+  "Volumes": {},
+  "User": "",
+  "WorkingDir": "",
+  "Hostname": "",
+  "ExposedPorts": {},
+  "NetworkingConfig": {},
+  "Healthcheck": {},
+  "HostConfig": {}
+}
+```
+
+空列表或空 map 字段有的会保留，有的会被省略到 `{}`，具体见下面映射规则。
+
+## 8. create.config 到 Docker JSON
+
+| 中间字段 | Docker JSON 字段 | 转换规则 |
+| --- | --- | --- |
+| `image` | `Image` | 转成 binary；缺省为 `""`。部署流程拉镜像时会对 image 补 `:latest`，但 create JSON 使用传入值。 |
+| `cmd` | `Cmd` | 列表元素逐个转 binary。 |
+| `entrypoint` | `Entrypoint` | 列表元素逐个转 binary。 |
+| `env` | `Env` | 列表元素逐个转 binary，并自动注入 `CONTAINER_NAME=...`。 |
+| `labels` | `Labels` | key/value 都转 binary。 |
+| `volumes` | `Volumes` | 转成 Docker 要求的 object，key 是容器内路径，value 是 `{}`。 |
+| `user` | `User` | 转成 binary；缺省为 `""`。 |
+| `working_dir` | `WorkingDir` | 转成 binary；缺省为 `""`。 |
+| `hostname` | `Hostname` | 转成 binary；缺省为 `""`。 |
+| `exposed_ports` | `ExposedPorts` | 转成 Docker 端口 object。 |
+| `healthcheck` | `Healthcheck` | 转成 Docker Healthcheck object；未传时为 `{}`。 |
+
+### Volumes
+
+输入：
+
+```erlang
+[<<"/data">>, <<"/usr/local/etc/service.conf">>]
+```
+
+输出 JSON：
+
+```json
+{
+  "Volumes": {
+    "/data": {},
+    "/usr/local/etc/service.conf": {}
+  }
+}
+```
+
+### ExposedPorts
+
+输入：
+
+```erlang
+[
+    #{container_port => 80, protocol => <<"tcp">>},
+    #{container_port => 53, protocol => <<"udp">>}
+]
+```
+
+输出 JSON：
+
+```json
+{
+  "ExposedPorts": {
+    "80/tcp": {},
+    "53/udp": {}
+  }
+}
+```
+
+规则：
+
+- `protocol` 为空或 `<<"tcp">>` 时输出 `tcp`。
+- 其他 protocol 原样输出。
+
+### Healthcheck
+
+输入：
+
+```erlang
+#{
+    test => [<<"CMD-SHELL">>, <<"curl -f http://localhost || exit 1">>],
+    interval_ns => 30000000000,
+    timeout_ns => 10000000000,
+    retries => 3
+}
+```
+
+输出 JSON：
+
+```json
+{
+  "Healthcheck": {
+    "Test": ["CMD-SHELL", "curl -f http://localhost || exit 1"],
+    "Interval": 30000000000,
+    "Timeout": 10000000000,
+    "Retries": 3
+  }
+}
+```
+
+## 9. create.host_config 到 Docker JSON
+
+`host_config` 会被合并到 Docker JSON 的 `HostConfig` 字段。
+
+| 中间字段 | Docker JSON 字段 | 转换规则 |
+| --- | --- | --- |
+| `binds` | `HostConfig.Binds` | 列表元素逐个转 binary，并自动注入 `service.conf` bind。空列表时省略。 |
+| `network_mode` | `HostConfig.NetworkMode` | 空 binary 时省略。 |
+| `restart_policy` | `HostConfig.RestartPolicy` | 转成 `Name` 和可选 `MaximumRetryCount`。 |
+| `privileged` | `HostConfig.Privileged` | 只有 true 时输出。false 时省略。 |
+| `cap_add` / `cap_drop` | `HostConfig.CapAdd` / `HostConfig.CapDrop` | 两者都为空时省略；只要一个非空，两个字段都会输出。 |
+| `devices` | `HostConfig.Devices` | 转成 Docker device object 列表。 |
+| `memory` | `HostConfig.Memory` | 0 时省略。 |
+| `memory_reservation` | `HostConfig.MemoryReservation` | 0 时省略。 |
+| `nano_cpus` | `HostConfig.NanoCpus` | 0 时省略。 |
+| `cpu_shares` | `HostConfig.CpuShares` | 0 时省略。 |
+| `ulimits` | `HostConfig.Ulimits` | 空列表时省略。 |
+| `tmpfs` | `HostConfig.Tmpfs` | 空 map 时省略。 |
+| `sysctls` | `HostConfig.Sysctls` | 空 map 时省略。 |
+| `extra_hosts` | `HostConfig.ExtraHosts` | 空列表时省略。 |
+
+### Binds
+
+输入：
+
+```erlang
+[
+    <<"/host/data:/data">>,
+    <<"/host/log:/var/log:ro">>
+]
+```
+
+efka 自动补丁后输出：
+
+```json
+{
+  "HostConfig": {
+    "Binds": [
+      "/path/to/container/service.conf:/usr/local/etc/service.conf",
+      "/host/data:/data",
+      "/host/log:/var/log:ro"
+    ]
+  }
+}
+```
+
+### RestartPolicy
+
+输入：
+
+```erlang
+#{name => <<"always">>, maximum_retry_count => 0}
+```
+
+输出：
+
+```json
+{
+  "HostConfig": {
+    "RestartPolicy": {
+      "Name": "always"
+    }
+  }
+}
+```
+
+输入：
+
+```erlang
+#{name => <<"on-failure">>, maximum_retry_count => 3}
+```
+
+输出：
+
+```json
+{
+  "HostConfig": {
+    "RestartPolicy": {
+      "Name": "on-failure",
+      "MaximumRetryCount": 3
+    }
+  }
+}
+```
+
+### Devices
+
+输入：
+
+```erlang
+[
+    #{
+        path_on_host => <<"/dev/ttyUSB0">>,
+        path_in_container => <<"/dev/ttyUSB0">>,
+        cgroup_permissions => <<"rwm">>
+    }
+]
+```
+
+输出：
+
+```json
+{
+  "HostConfig": {
+    "Devices": [
+      {
+        "PathOnHost": "/dev/ttyUSB0",
+        "PathInContainer": "/dev/ttyUSB0",
+        "CgroupPermissions": "rwm"
+      }
+    ]
+  }
+}
+```
+
+`cgroup_permissions` 为空时默认输出 `rwm`。
+
+### 资源限制
+
+输入：
+
+```erlang
+#{
+    memory => 536870912,
+    memory_reservation => 268435456,
+    nano_cpus => 1500000000,
+    cpu_shares => 512
+}
+```
+
+输出：
+
+```json
+{
+  "HostConfig": {
+    "Memory": 536870912,
+    "MemoryReservation": 268435456,
+    "NanoCpus": 1500000000,
+    "CpuShares": 512
+  }
+}
+```
+
+值为 0 的资源字段会被省略。
+
+### Ulimits
+
+输入：
+
+```erlang
+[
+    #{name => <<"nofile">>, soft => 1024, hard => 2048}
+]
+```
+
+输出：
+
+```json
+{
+  "HostConfig": {
+    "Ulimits": [
+      {
+        "Name": "nofile",
+        "Soft": 1024,
+        "Hard": 2048
+      }
+    ]
+  }
+}
+```
+
+### Tmpfs
+
+输入：
+
+```erlang
+#{
+    <<"/tmp">> => <<>>,
+    <<"/run">> => <<"rw,size=64m">>
+}
+```
+
+输出：
+
+```json
+{
+  "HostConfig": {
+    "Tmpfs": {
+      "/tmp": "",
+      "/run": "rw,size=64m"
+    }
+  }
+}
+```
+
+### Sysctls
+
+输入：
+
+```erlang
+#{<<"net.ipv4.ip_forward">> => <<"1">>}
+```
+
+输出：
+
+```json
+{
+  "HostConfig": {
+    "Sysctls": {
+      "net.ipv4.ip_forward": "1"
+    }
+  }
+}
+```
+
+### ExtraHosts
+
+输入：
+
+```erlang
+[<<"host.docker.internal:host-gateway">>]
+```
+
+输出：
+
+```json
+{
+  "HostConfig": {
+    "ExtraHosts": ["host.docker.internal:host-gateway"]
+  }
+}
+```
+
+## 10. create.networking_config 到 Docker JSON
+
+输入：
+
+```erlang
+#{
+    endpoints => [
+        #{name => <<"bridge">>},
+        #{name => <<"mynet">>}
+    ]
+}
+```
+
+输出：
+
+```json
+{
+  "NetworkingConfig": {
+    "EndpointsConfig": {
+      "bridge": {},
+      "mynet": {}
+    }
+  }
+}
+```
+
+未传或 endpoints 为空时输出：
+
+```json
+{
+  "NetworkingConfig": {}
+}
+```
+
+## 11. 完整转换示例
+
+收到的 deploy command：
+
+```erlang
+{command, Ref, {container, #{
+    action => deploy,
+    task_id => 1001,
+    params => #{
+        container_name => <<"my_nginx">>,
+        container_dir => <<"/data/apps/my_nginx">>,
+        create => #{
+            config => #{
+                image => <<"docker.io/library/nginx:latest">>,
+                cmd => [<<"nginx">>, <<"-g">>, <<"daemon off;">>],
+                env => [<<"ENV=prod">>],
+                volumes => [<<"/data">>],
+                exposed_ports => [#{container_port => 80, protocol => <<"tcp">>}]
+            },
+            host_config => #{
+                binds => [<<"/host/data:/data">>],
+                restart_policy => #{name => <<"always">>, maximum_retry_count => 0},
+                memory => 536870912
+            },
+            networking_config => #{
+                endpoints => [#{name => <<"bridge">>}]
+            }
+        }
+    }
+}}}
+```
+
+生成的 Docker JSON 示意：
+
+```json
+{
+  "Image": "docker.io/library/nginx:latest",
+  "Cmd": ["nginx", "-g", "daemon off;"],
+  "Entrypoint": [],
+  "Env": ["CONTAINER_NAME=my_nginx", "ENV=prod"],
+  "Labels": {},
+  "Volumes": {
+    "/usr/local/etc/service.conf": {},
+    "/data": {}
+  },
+  "User": "",
+  "WorkingDir": "",
+  "Hostname": "",
+  "ExposedPorts": {
+    "80/tcp": {}
+  },
+  "NetworkingConfig": {
+    "EndpointsConfig": {
+      "bridge": {}
+    }
+  },
+  "Healthcheck": {},
+  "HostConfig": {
+    "Binds": [
+      "/data/apps/my_nginx/service.conf:/usr/local/etc/service.conf",
+      "/host/data:/data"
+    ],
+    "RestartPolicy": {
+      "Name": "always"
+    },
+    "Memory": 536870912
+  }
+}
+```
+
+## 12. 部署流程补充
+
+deploy action 的执行流程：
+
+1. 根据 `container_name` 和 `container_dir` 确保容器目录存在。
+2. 启动独立部署进程。
+3. 部署进程上报任务事件流。
+4. 规范化镜像名：如果镜像最后一段没有 tag，则补 `:latest`。
+5. 调用 Docker API 拉取镜像。
+6. 构造 Docker create JSON。
+7. 调用 `POST /containers/create?name={container_name}`。
+8. 创建空 `service.conf` 文件。
+9. 写入部署摘要日志并关闭任务事件流。
+
+注意：当前 `ensure_container_absent/2` 只上报“开始创建容器”，不会删除已有同名容器；如果 Docker 返回名称冲突，部署会失败并上报“本地容器已经存在”。
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/transport/efka_client.erl b/src/transport/efka_client.erl
index 696b24c..e878169 100644
--- a/src/transport/efka_client.erl
+++ b/src/transport/efka_client.erl
@@ -34,6 +34,8 @@
     socket :: undefined | ssl:sslsocket(),
     %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
     auth_ref = undefined :: undefined | reference(),
+    %% iot auth 命令触发重新鉴权时，用它保存 command 的 ref。
+    auth_control_ref = undefined :: undefined | reference(),
     dropped_message_count = 0 :: non_neg_integer()
 }).
 
@@ -137,11 +139,12 @@ handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State =
             {keep_state, State#state{socket = undefined}}
     end;
 
-handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = Socket}) ->
+handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = Socket, auth_control_ref = AuthControlRef}) ->
     logger:debug("[efka_client] auth request timeout"),
+    maybe_send_auth_response(Socket, AuthControlRef, {error, timeout}),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}};
 
 %% 将缓存中的数据推送到服务器端
 handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
@@ -164,87 +167,95 @@ handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socke
     logger:debug("[efka_client] ssl error: ~p", [Reason]),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}};
 handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) ->
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}};
 
 %%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配
 
-%% 容器管理请求
-handle_event(internal, {request, Ref, {container_request, #{action := list}}},
+%% 容器管理命令由 iot 发起，使用 command/command_response 语义。
+handle_event(internal, {command, Ref, {container, #{action := list}}},
     ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_commands:get_containers(),
-    handle_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {request, Ref, {container_request, #{action := deploy, task_id := TaskId, params := Params}}},
+handle_event(internal, {command, Ref, {container, #{action := deploy, task_id := TaskId, params := Params}}},
     ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_deploy_manager:deploy(TaskId, Params),
-    handle_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {request, Ref, {container_request, #{action := start, target := Target}}},
+handle_event(internal, {command, Ref, {container, #{action := start, target := Target}}},
     ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_commands:start_container(container_target(Target)),
-    handle_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {request, Ref, {container_request, #{action := stop, target := Target, timeout_seconds := TimeoutSeconds}}},
+handle_event(internal, {command, Ref, {container, #{action := stop, target := Target, timeout_seconds := TimeoutSeconds}}},
     ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
-    handle_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {request, Ref, {container_request, #{action := kill, target := Target, signal := Signal}}},
+handle_event(internal, {command, Ref, {container, #{action := kill, target := Target, signal := Signal}}},
     ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
-    handle_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {request, Ref, {container_request, #{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}}},
+handle_event(internal, {command, Ref, {container, #{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}}},
     ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
-    handle_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {request, Ref, {container_request, #{action := config, target := Target, config := Config}}},
+handle_event(internal, {command, Ref, {container, #{action := config, target := Target, config := Config}}},
     ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
-    handle_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {request, Ref, {container_request, Request}}, ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
-    logger:notice("[efka_client] get a invalid request: ~p, agent restricted", [Request]),
-    handle_container_response(Socket, Ref, {error, <<"agent restricted">>}),
+handle_event(internal, {command, Ref, {container, Request}}, ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
+    logger:notice("[efka_client] get an invalid command: ~p, agent restricted", [Request]),
+    send_container_response(Socket, Ref, {error, <<"agent restricted">>}),
     {keep_state, State};
-handle_event(internal, {request, Ref, {container_request, Request}}, _StateName, State = #state{socket = Socket}) ->
-    logger:notice("[efka_client] get a invalid request: ~p, agent restricted", [Request]),
-    handle_container_response(Socket, Ref, {error, <<"agent invalid">>}),
+handle_event(internal, {command, Ref, {container, Request}}, _StateName, State = #state{socket = Socket}) ->
+    logger:notice("[efka_client] get an invalid command: ~p, agent invalid", [Request]),
+    send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
     {keep_state, State};
 
 %% 处理response
-handle_event(internal, {response, AuthRef, {auth_response, ok}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
+handle_event(internal, {response, AuthRef, {auth_response, ok}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef, auth_control_ref = AuthControlRef}) ->
     logger:debug("[efka_client] auth success"),
-    {next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined}, [{next_event, info, flush_cache}]};
-handle_event(internal, {response, AuthRef, {auth_response, {error, {denied, Message}}}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
+    maybe_send_auth_response(Socket, AuthControlRef, ok),
+    {next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined, auth_control_ref = undefined}, [{next_event, info, flush_cache}]};
+handle_event(internal, {response, AuthRef, {auth_response, {error, {denied, Message}}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef, auth_control_ref = AuthControlRef}) ->
     logger:debug("[efka_client] auth denied, message: ~p", [Message]),
-    {next_state, ?STATE_RESTRICTED, State#state{auth_ref = undefined}};
-handle_event(internal, {response, AuthRef, {auth_response, {error, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
+    maybe_send_auth_response(Socket, AuthControlRef, {error, {denied, Message}}),
+    {next_state, ?STATE_RESTRICTED, State#state{auth_ref = undefined, auth_control_ref = undefined}};
+handle_event(internal, {response, AuthRef, {auth_response, {error, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef, auth_control_ref = AuthControlRef}) ->
     logger:debug("[efka_client] auth failed, reason: ~p", [Reason]),
+    maybe_send_auth_response(Socket, AuthControlRef, {error, Reason}),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}};
 handle_event(internal, {response, _Ref, Reply}, StateName, State) ->
     logger:warning("[efka_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
+handle_event(internal, {command_response, _Ref, Reply}, StateName, State) ->
+    logger:warning("[efka_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]),
+    {keep_state, State};
 
 %% 处理命令
-handle_event(internal, {message, {auth_control, Cmd}}, StateName, State = #state{socket = Socket}) ->
+handle_event(internal, {command, CommandRef, {auth, Cmd}}, StateName, State = #state{socket = Socket}) ->
     logger:debug("[efka_client] auth cmd: ~p", [Cmd]),
     case {Cmd, StateName} of
         {activate, ?STATE_ACTIVATED} ->
+            send_auth_response(Socket, CommandRef, ok),
             {keep_state, State};
         {activate, _} ->
             Ref = make_ref(),
             AuthPacket = auth_packet(Ref),
             ok = ssl:send(Socket, AuthPacket),
-            {next_state, ?STATE_AUTH, State#state{auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
+            {next_state, ?STATE_AUTH, State#state{auth_ref = Ref, auth_control_ref = CommandRef}, [{state_timeout, 5000, auth_timeout}]};
         {deactivate, _} ->
-            {next_state, ?STATE_RESTRICTED, State}
+            send_auth_response(Socket, CommandRef, ok),
+            {next_state, ?STATE_RESTRICTED, State#state{auth_ref = undefined, auth_control_ref = undefined}}
     end;
 
 %% 处理Pub/Sub机制
@@ -311,11 +322,22 @@ disconnect(Socket) ->
 schedule_reconnect() ->
     erlang:start_timer(5000, self(), create_transport).
 
--spec handle_container_response(ssl:sslsocket(), reference(), term()) -> ok.
-handle_container_response(Socket, Ref, Reply) ->
-    Packet = term_to_binary({response, Ref, {container_response, Reply}}),
+-spec send_container_response(ssl:sslsocket(), reference(), term()) -> ok.
+send_container_response(Socket, Ref, Reply) ->
+    Packet = term_to_binary({command_response, Ref, {container, Reply}}),
     ok = ssl:send(Socket, Packet).
 
+-spec send_auth_response(ssl:sslsocket(), reference(), term()) -> ok.
+send_auth_response(Socket, Ref, Reply) ->
+    Packet = term_to_binary({command_response, Ref, {auth, Reply}}),
+    ok = ssl:send(Socket, Packet).
+
+-spec maybe_send_auth_response(ssl:sslsocket(), undefined | reference(), term()) -> ok.
+maybe_send_auth_response(_Socket, undefined, _Reply) ->
+    ok;
+maybe_send_auth_response(Socket, Ref, Reply) when is_reference(Ref) ->
+    send_auth_response(Socket, Ref, Reply).
+
 -spec container_target(map()) -> binary().
 container_target(Target) when is_map(Target) ->
     NameBin = to_binary(maps:get(name, Target, <<>>)),
@@ -342,4 +364,4 @@ to_bool(1) ->
 to_bool(false) ->
     false;
 to_bool(0) ->
-    false.
\ No newline at end of file
+    false.