chore: optimize Dockerfile apt source compatibility and update edge.yaml cloud/ota URLs to port 9000

- Dockerfile: support both legacy sources.list and DEB822 format for apt mirror - Dockerfile: use official python:3.9-slim instead of Huawei Cloud SWR mirror - .dockerignore: exclude !build/edge-agent to avoid build artifact in context - edge.yaml: change cloud_url from 8004 to 9000 (stream-gateway) - edge.yaml: change ota_url from 8087 to 9000 - Add Dockerfile.bak and buildkitd.toml
fix(ota): disable elf check and fix self-update binary replacement
2026-05-09 21:59:11 +08:00 · 2026-05-09 14:16:27 +08:00 · 2026-05-09 13:50:07 +08:00 · 2026-05-09 13:36:12 +08:00 · 2026-05-09 11:19:05 +08:00 · 2026-05-09 11:03:27 +08:00
33 changed files with 2125 additions and 153 deletions
--- a/.dockerignore
+++ b/.dockerignore
@@ -0,0 +1,12 @@
+.git/
+.gitignore
+build/
+!build/edge-agent
+data/
+*.om
+*.onnx
+*.tar.gz
+logs/
+venv/
+__pycache__/
+*.pyc
--- a/.gitignore
+++ b/.gitignore
@@ -0,0 +1,7 @@
+build/
+logs/
+*.om
+*.onnx
+test.jpg
+fusion_result.json
+device.uuid
--- a/DEPLOYMENT.md
+++ b/DEPLOYMENT.md
@@ -0,0 +1,288 @@
+# AI-tianyan 边缘部署操作指南
+
+## 系统环境
+
+| 项目 | 要求 |
+|---|---|
+| 硬件 | Atlas 200I A2 (Ascend 310B4) 或同系列设备 |
+| OS | Ubuntu 22.04 aarch64 |
+| Python | 3.9.x (Miniconda 或系统自带) |
+| CANN | 6.2.RC2 (Ascend Toolkit V100R003C11) |
+| Go | 1.18+ (仅编译时需要) |
+| 云端 | 101.36.73.102 (API/MQTT/InfluxDB/OTA/ZLMediaKit) |
+
+> 如果目标设备环境完全一致（同镜像版本），可直接复用已编译的 Go 二进制和 .om 模型，无需重新编译。
+
+---
+
+## 部署方式
+
+### 方式一：离线包部署（推荐，最快）
+
+适用于环境一致的设备，所有构建产物已包含在内。
+
+#### 1. 在源设备打包
+
+```bash
+cd /root
+tar czf ai-tianyan.tar.gz --exclude='.git' --exclude='__pycache__' --exclude='*.pyc' AI-tianyan/
+```
+
+#### 2. 传输到目标设备
+
+```bash
+# 网络传输
+scp ai-tianyan.tar.gz root@目标IP:/root/
+
+# 或 U盘拷贝
+```
+
+#### 3. 目标设备上安装
+
+```bash
+# 确保 pip 存在
+python3 -m pip --version || apt update && apt install -y python3-pip
+
+# 解压
+cd /root
+tar xzf ai-tianyan.tar.gz
+cd AI-tianyan
+
+# 一键安装（部署文件 + Python依赖 + systemd服务 + 自动启动）
+bash scripts/install.sh
+```
+
+#### 4. 修改配置
+
+```bash
+# 生成新的设备 UUID
+NEW_UUID=$(cat /proc/sys/kernel/random/uuid | tr -d '-')
+
+# 更新配置
+sudo sed -i "s/device_uuid:.*/device_uuid: $NEW_UUID/" /opt/tianyan-edge/config/edge.yaml
+sudo sed -i "s/edge_id:.*/edge_id: edge-demo-002/" /opt/tianyan-edge/config/edge.yaml
+
+# 如有不同的摄像头流地址
+sudo vi /opt/tianyan-edge/config/edge.yaml
+```
+
+#### 5. 验证
+
+```bash
+sudo systemctl status edge-agent edge-infer
+journalctl -u edge-agent -n 50 --no-pager
+```
+
+---
+
+### 方式二：Git 源码部署
+
+适用于需要自定义代码或环境有差异的设备。
+
+#### 1. 拉取代码
+
+```bash
+git clone http://101.36.73.102:3112/fyah/AI-tianyan.git
+cd AI-tianyan
+```
+
+#### 2. 编译 Go 边缘代理
+
+```bash
+export GOPROXY=https://goproxy.cn,direct
+bash scripts/build.sh
+# 输出: build/edge-agent
+```
+
+#### 3. 转换模型（仅 ATC 版本不同时需要）
+
+```bash
+source /usr/local/Ascend/ascend-toolkit/set_env.sh
+bash scripts/atc_convert.sh
+# 或手动:
+atc --model=yolov8n.onnx \
+    --framework=5 \
+    --output=model/model.om \
+    --soc_version=Ascend310B4 \
+    --input_format=NCHW \
+    --input_shape="images:1,3,640,640" \
+    --output_type=FP16
+```
+
+#### 4. 安装部署
+
+```bash
+# 确保 pip 存在
+python3 -m pip --version || apt update && apt install -y python3-pip
+
+# 安装 Python 依赖
+python3 -m pip install opencv-python-headless numpy
+
+# 一键安装
+bash scripts/install.sh
+```
+
+---
+
+## 安装脚本说明
+
+`scripts/install.sh` 会自动执行以下操作：
+
+```
+1. 创建目录: /opt/tianyan-edge/{bin,python,config,systemd,staging,model}
+2. 复制 edge-agent -> /opt/tianyan-edge/bin/edge-agent
+3. 复制 infer_server.py -> /opt/tianyan-edge/python/
+4. 复制 edge.yaml.template -> /opt/tianyan-edge/config/edge.yaml (仅在不存在时)
+5. 安装 Python 依赖 (opencv-python-headless, numpy)
+6. 安装 systemd 服务文件
+7. daemon-reload + enable + restart 服务
+```
+
+---
+
+## 配置文件说明
+
+安装后配置位于 `/opt/tianyan-edge/config/edge.yaml`：
+
+```yaml
+device_uuid: 8541db9f77826e39605ef2c032f8fb93   # 设备唯一标识，每台设备必须不同
+edge_id: edge-demo-001                             # 设备名称，便于识别
+cloud_url: http://101.36.73.102:8004               # 云端 API 地址
+mqtt_broker: tcp://101.36.73.102:1883              # MQTT Broker
+mqtt_user: ""                                      # MQTT 认证用户名
+mqtt_pass: ""                                      # MQTT 认证密码
+edge_token: ""                                     # 设备认证 Token
+rtsp_urls:
+  - http://101.36.73.102:8080/rtp/xxx.live.flv     # 视频流地址 (支持多路)
+infer_socket: /tmp/edge-infer.sock                 # Go与Python通信的Unix Socket
+infer_fps: 2                                       # 推理帧率 (每秒抽样数)
+infer_workers: 3                                   # 并发推理 worker 数
+conf_threshold: 0.2                                # 检测置信度阈值
+dedup_window_sec: 30                               # 事件去重窗口 (秒)
+ota_url: http://101.36.73.102:8087                 # OTA 自动更新地址
+version: 1.0.0                                     # 固件版本号
+```
+
+> 新设备部署时 **必须修改**: `device_uuid`、`edge_id`、`rtsp_urls`
+
+---
+
+## 服务管理
+
+### 启动/停止/重启
+
+```bash
+sudo systemctl start edge-infer          # 启动 NPU 推理服务
+sudo systemctl start edge-agent          # 启动边缘代理
+sudo systemctl stop edge-infer edge-agent
+sudo systemctl restart edge-infer edge-agent
+```
+
+### 开机自启
+
+```bash
+sudo systemctl enable edge-infer edge-agent
+```
+
+### 查看状态
+
+```bash
+sudo systemctl status edge-agent edge-infer
+sudo journalctl -u edge-agent -f         # 实时查看边缘代理日志
+sudo journalctl -u edge-infer -f         # 实时查看推理服务日志
+```
+
+### 手动调试模式
+
+```bash
+# 停止 systemd 服务
+sudo systemctl stop edge-agent edge-infer
+
+# 手动启动推理服务
+bash -lc 'source /usr/local/Ascend/ascend-toolkit/set_env.sh && \
+  NAMES_FILE=/opt/tianyan-edge/model/names.txt \
+  OM_MODEL=/opt/tianyan-edge/model/model.om \
+  CONF_THRESHOLD=0.2 \
+  OUTPUT_FORMAT=raw \
+  python3 /opt/tianyan-edge/python/infer_server.py'
+
+# 另一个终端启动边缘代理
+/opt/tianyan-edge/bin/edge-agent -config /opt/tianyan-edge/config/edge.yaml
+```
+
+---
+
+## 运行验证
+
+```bash
+# 检查进程
+ps aux | grep -E 'edge-agent|infer_server'
+
+# 检查 NPU 状态
+npu-smi info
+
+# 检查网络连接 (应看到连接到 101.36.73.102)
+ss -tunp | grep 101.36.73.102
+
+# 检查 Unix Socket
+ls -la /tmp/edge-infer.sock
+
+# 检查模型文件
+ls -lh /opt/tianyan-edge/model/model.om /opt/tianyan-edge/model/names.txt
+
+# 查看系统日志
+tail -f /opt/tianyan-edge/logs/agent.log 2>/dev/null
+tail -f /opt/tianyan-edge/logs/infer.log 2>/dev/null
+```
+
+---
+
+## Docker 部署（可选）
+
+```bash
+cd AI-tianyan
+docker compose up -d
+docker compose logs -f tianyan-edge
+```
+
+要求：Docker 已安装，CANN 驱动已就绪，NPU 设备节点 `/dev/davinci0` 等存在。
+
+---
+
+## 常见问题
+
+| 问题 | 解决方法 |
+|---|---|
+| `acl.init failed` | 确认 `source set_env.sh` 已执行，检查 `LD_LIBRARY_PATH` 包含 CANN 路径 |
+| `model.om not found` | 确认 `/opt/tianyan-edge/model/model.om` 存在，重新执行 ATC 转换 |
+| `python3 -m pip: command not found` | `apt install -y python3-pip` |
+| `opencv-python-headless 安装失败` | `apt install -y python3-dev gcc g++ libgl1-mesa-glx` 后再 pip install |
+| 无法连接 MQTT | 检查 `mqtt_broker` 地址是否正确，确认网络可达 |
+| 拉流失败 | 确认 ZLMediaKit 端口 8080 可达，流地址正确 |
+| Go 编译失败 | `export GOPROXY=https://goproxy.cn,direct && go mod download` |
+| systemd 启动后立即退出 | `journalctl -u edge-agent -n 100` 查看详细错误 |
+| 多设备 UUID 冲突 | 每台设备必须有不同的 `device_uuid`，用 `cat /proc/sys/kernel/random/uuid` 生成 |
+
+---
+
+## 卸载
+
+```bash
+bash scripts/uninstall.sh
+# 或 make uninstall
+```
+
+---
+
+## OTA 远程更新
+
+云端推送新版本后，edge-agent 会自动检测并下载更新到 staging 目录。
+也可手动更新：
+
+```bash
+cd /root/AI-tianyan
+git pull
+make build
+make install
+sudo systemctl restart edge-agent edge-infer
+```
--- a/44
+++ b/44
@@ -0,0 +1,44 @@
+# 使用华为云镜像加速的 Python 3.9 Slim
+FROM python:3.9-slim
+
+ENV DEBIAN_FRONTEND=noninteractive
+# 替换 apt 源为华为云镜像（兼容 Debian 11 旧格式和 Debian 12/13 DEB822 格式）
+RUN if [ -f /etc/apt/sources.list ]; then \
+      sed -i 's/deb.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list && \
+      sed -i 's/security.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list; \
+    fi && \
+    if [ -f /etc/apt/sources.list.d/debian.sources ]; then \
+      sed -i 's|https\?://deb.debian.org|http://repo.huaweicloud.com|g' /etc/apt/sources.list.d/debian.sources && \
+      sed -i 's|https\?://security.debian.org|http://repo.huaweicloud.com|g' /etc/apt/sources.list.d/debian.sources; \
+    fi && \
+    apt-get update && apt-get install -y --no-install-recommends \
+    ffmpeg curl libgl1 libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /opt/tianyan
+
+# 1. 复制已编译的 Go 二进制 (由宿主机构建)
+COPY build/edge-agent /opt/tianyan/edge-agent
+
+# 2. 安装 Python 依赖 (使用华为云 pip 源)
+COPY python/requirements.txt .
+RUN pip3 install --no-cache-dir -r requirements.txt -i https://repo.huaweicloud.com/repository/pypi/simple \
+    && rm -rf /root/.cache/pip
+
+# 3. 复制业务代码与脚本
+COPY python/ /opt/tianyan/python/
+COPY scripts/ /opt/tianyan/scripts/
+
+# 4. 创建运行时目录
+RUN mkdir -p /opt/tianyan/config /opt/tianyan/model /tmp
+
+# 环境变量 (CANN 路径将由 docker-compose 挂载覆盖)
+ENV PYTHONPATH=/usr/local/Ascend/ascend-toolkit/latest/python/site-packages
+ENV LD_LIBRARY_PATH=/usr/local/Ascend/ascend-toolkit/latest/lib64:/usr/local/Ascend/driver/lib64
+ENV PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
+
+# 启动脚本
+COPY entrypoint.sh /opt/tianyan/entrypoint.sh
+RUN chmod +x /opt/tianyan/entrypoint.sh
+
+ENTRYPOINT ["/opt/tianyan/entrypoint.sh"]
--- a/Dockerfile.bak
+++ b/Dockerfile.bak
@@ -0,0 +1,38 @@
+# 使用华为云镜像加速的 Python 3.9 Slim
+FROM swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/library/python:3.9-slim
+
+ENV DEBIAN_FRONTEND=noninteractive
+# 安装 ffmpeg, curl 及 OpenCV 基础依赖 (使用华为云 apt 源)
+RUN sed -i 's/deb.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list && \
+    sed -i 's/security.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list && \
+    apt-get update && apt-get install -y --no-install-recommends \
+    ffmpeg curl libgl1 libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /opt/tianyan
+
+# 1. 复制已编译的 Go 二进制 (由宿主机构建)
+COPY build/edge-agent /opt/tianyan/edge-agent
+
+# 2. 安装 Python 依赖 (使用华为云 pip 源)
+COPY python/requirements.txt .
+RUN pip3 install --no-cache-dir -r requirements.txt -i https://repo.huaweicloud.com/repository/pypi/simple \
+    && rm -rf /root/.cache/pip
+
+# 3. 复制业务代码与脚本
+COPY python/ /opt/tianyan/python/
+COPY scripts/ /opt/tianyan/scripts/
+
+# 4. 创建运行时目录
+RUN mkdir -p /opt/tianyan/config /opt/tianyan/model /tmp
+
+# 环境变量 (CANN 路径将由 docker-compose 挂载覆盖)
+ENV PYTHONPATH=/usr/local/Ascend/ascend-toolkit/latest/python/site-packages
+ENV LD_LIBRARY_PATH=/usr/local/Ascend/ascend-toolkit/latest/lib64:/usr/local/Ascend/driver/lib64
+ENV PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
+
+# 启动脚本
+COPY entrypoint.sh /opt/tianyan/entrypoint.sh
+RUN chmod +x /opt/tianyan/entrypoint.sh
+
+ENTRYPOINT ["/opt/tianyan/entrypoint.sh"]
--- a/buildkitd.toml
+++ b/buildkitd.toml
@@ -0,0 +1,2 @@
+[registry."docker.io"]
+  mirrors = ["docker.1panel.live", "docker.1panel.dev", "docker.1ms.run"]
--- a/cmd/edge-agent/main.go
+++ b/cmd/edge-agent/main.go
@@ -21,13 +21,14 @@ func main() {
 	flag.Parse()

 	cfg := config.Load(*cfgPath)
-	log.Printf("edge-agent start id=%s cloud=%s", cfg.EdgeID, cfg.CloudURL)
+	log.Printf("edge-agent start id=%s uuid=%s cloud=%s mqtt=%s auto_pull=%v",
+		cfg.EdgeID, cfg.GetDeviceIdentity(), cfg.CloudURL, cfg.MqttBroker, cfg.AutoPull)

 	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer cancel()

-	frames := make(chan stream.Frame, 50)
-	events := make(chan event.SuspectedEvent, 200)
+	frames := make(chan stream.Frame, 100)
+	events := make(chan event.SuspectedEvent, 500)

 	var wg sync.WaitGroup
 	run := func(fn func(context.Context)) {
@@ -38,11 +39,32 @@ func main() {
 		}()
 	}

-	run(stream.NewIngestor(cfg, frames).Run)
-	run(infer.NewClient(cfg, frames, events).Run)
+	// 1. Cloud Communication
+	mqttMgr := control.NewMqttManager(cfg)
+	run(mqttMgr.Run)
+
+	// 1.5 Stream Puller (Optional: Dynamic URL fetching)
+	var puller *stream.Puller
+	if cfg.AutoPull {
+		puller = stream.NewPuller(cfg)
+		puller.StartHeartbeatLoop()
+		defer puller.Stop()
+		log.Println("puller: auto-pull mode enabled")
+	}
+
+	// 2. Stream Ingestion (Dynamic)
+	run(stream.NewStreamManager(cfg, puller, frames, mqttMgr.GetUpdates()).Run)
+
+	// 3. Inference (Dynamic Workers)
+	run(infer.NewClient(cfg, frames, events, mqttMgr.GetUpdates()).Run)
+
+	// 4. Event Upload
 	run(event.NewUploader(cfg, events).Run)
+
+	// 5. Heartbeat
 	run(control.NewHeartbeat(cfg).Run)
-	run(control.NewConfigAgent(cfg).Run)
+
+	// 6. OTA Update
 	run(control.NewOTAAgent(cfg).Run)

 	wg.Wait()
--- a/config/edge.yaml
+++ b/config/edge.yaml
@@ -1,13 +1,19 @@
+device_uuid: 8541db9f77826e39605ef2c032f8fb93
 edge_id: edge-demo-001
-cloud_url: http://101.36.73.102:8004
-edge_token: ""
-
-# RTSP 流地址（留空使用演示模式）
+cloud_url: http://101.36.73.102:9000
+mqtt_broker: tcp://101.36.73.102:1883
+mqtt_user: ""
+mqtt_pass: ""
+edge_token: "edge-token-001"
 rtsp_urls: []
-
+# 自动按需拉流配置
+stream_pull_url: http://101.36.73.102:9000
+stream_protocol: flv
+auto_pull: true
 infer_socket: /tmp/edge-infer.sock
-infer_fps: 5
+infer_fps: 2
 infer_workers: 3
-conf_threshold: 0.5
+conf_threshold: 0.2
 dedup_window_sec: 30
+ota_url: http://101.36.73.102:9000
 version: 1.0.0
--- a/config/telegraf.conf
+++ b/config/telegraf.conf
@@ -0,0 +1,32 @@
+[agent]
+  interval = "10s"
+  round_interval = true
+  metric_batch_size = 1000
+  metric_buffer_limit = 10000
+  flush_interval = "10s"
+  hostname = "edge-001" # 启动时替换为实际 edge_id
+
+# 输出到云端 InfluxDB
+[[outputs.influxdb_v2]]
+  urls = ["http://101.36.73.102:18086"]
+  token = "my-super-secret-token"  # 替换为实际 token
+  organization = "tianyan"
+  bucket = "edge_metrics"
+
+# 基础硬件监控
+[[inputs.cpu]]
+  percpu = false
+  totalcpu = true
+
+[[inputs.mem]]
+
+[[inputs.disk]]
+  ignore_fs = ["tmpfs", "devtmpfs", "devfs", "overlay", "aufs", "squashfs"]
+
+[[inputs.net]]
+
+# NPU 自定义监控
+[[inputs.exec]]
+  commands = ["/opt/tianyan-edge/scripts/npu_info.sh"]
+  timeout = "5s"
+  data_format = "influx"
--- a/data/config/edge.yaml
+++ b/data/config/edge.yaml
@@ -0,0 +1,16 @@
+device_uuid: 8541db9f77826e39605ef2c032f8fb93
+edge_id: edge-demo-001
+cloud_url: http://101.36.73.102:9000
+mqtt_broker: tcp://101.36.73.102:1883
+mqtt_user: ""
+mqtt_pass: ""
+edge_token: "be079dc5ec8d7c3796d052f3c10143cd"
+rtsp_urls:
+    - "http://101.36.73.102:8080/rtp/34020000002000000003_34020000001310000001.live.flv?secret=be079dc5ec8d7c3796d052f3c10143cd"
+infer_socket: /tmp/edge-infer.sock
+infer_fps: 2
+infer_workers: 3
+conf_threshold: 0.2
+dedup_window_sec: 30
+ota_url: http://101.36.73.102:9000
+version: 1.0.0
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -0,0 +1,34 @@
+version: '3.8'
+
+services:
+  tianyan-edge:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    image: tianyan-edge:latest
+    container_name: tianyan-edge
+    restart: unless-stopped
+    network_mode: "host"
+    pid: "host"
+    
+    # 显式映射 NPU 设备
+    devices:
+      - /dev/davinci0
+      - /dev/davinci_manager
+      - /dev/devmm_svm
+      - /dev/hisi_hdc
+
+    volumes:
+      # 🔑 核心：挂载宿主机 CANN 驱动与库
+      - /usr/local/Ascend:/usr/local/Ascend:ro
+      
+      # 挂载配置与模型
+      - ./data/config:/opt/tianyan/config
+      - ./data/model:/opt/tianyan/model
+      
+      # 共享 Unix Socket 与日志
+      - /tmp:/tmp
+      - ./data/logs:/var/log/tianyan
+
+    environment:
+      - TZ=Asia/Shanghai
--- a/entrypoint.sh
+++ b/entrypoint.sh
@@ -0,0 +1,41 @@
+#!/bin/bash
+set -e
+
+echo "🚀 Starting Tianyan Edge Container..."
+
+# 1. 启动 NPU 推理服务 (后台)
+echo "[1/2] Starting infer_server.py..."
+cd /opt/tianyan
+python3 python/infer_server.py > /var/log/tianyan/infer.log 2>&1 &
+INFER_PID=$!
+
+# 2. 等待 Unix Socket 创建 (最多 20s)
+echo "⏳ Waiting for inference socket at $INFER_SOCKET..."
+for i in $(seq 1 20); do
+    if [ -S "${INFER_SOCKET:-/tmp/edge-infer.sock}" ]; then
+        echo "✅ Socket ready. Starting edge-agent..."
+        break
+    fi
+    sleep 1
+done
+
+if [ ! -S "${INFER_SOCKET:-/tmp/edge-infer.sock}" ]; then
+    echo "❌ ERROR: Inference socket not created. Check NPU/CANN status."
+    cat /var/log/tianyan/infer.log
+    kill $INFER_PID 2>/dev/null
+    exit 1
+fi
+
+# 3. 启动 Go 边缘代理
+/opt/tianyan/edge-agent -config /opt/tianyan/config/edge.yaml > /var/log/tianyan/agent.log 2>&1 &
+AGENT_PID=$!
+
+# 4. 信号捕获与优雅退出
+trap "echo 'Shutting down...'; kill $INFER_PID $AGENT_PID 2>/dev/null; wait; exit 0" SIGINT SIGTERM
+
+# 阻塞主进程
+wait -n
+EXIT_CODE=$?
+echo "Process exited with code $EXIT_CODE"
+kill $INFER_PID $AGENT_PID 2>/dev/null
+exit $EXIT_CODE
--- a/go.mod
+++ b/go.mod
@@ -1,7 +1,15 @@
 module tianyan-edge

-go 1.22
+go 1.18

-require gopkg.in/yaml.v3 v3.0.1
+require (
+	github.com/eclipse/paho.mqtt.golang v1.4.3
+	gopkg.in/yaml.v3 v3.0.1
+)

-require gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
+require (
+	github.com/gorilla/websocket v1.5.0 // indirect
+	golang.org/x/net v0.8.0 // indirect
+	golang.org/x/sync v0.1.0 // indirect
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
+)
--- a/go.sum
+++ b/go.sum
@@ -1,8 +1,16 @@
+github.com/eclipse/paho.mqtt.golang v1.4.3 h1:2kwcUGn8seMUfWndX0hGbvH8r7crgcJguQNCyp70xik=
+github.com/eclipse/paho.mqtt.golang v1.4.3/go.mod h1:CSYvoAlsMkhYOXh/oKyxa8EcBci6dVkLCbo5tTC1RIE=
+github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
+github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ=
+golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
+golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -1,27 +1,41 @@
 package config

 import (
+	"crypto/rand"
+	"encoding/hex"
 	"os"
+	"path/filepath"

 	"gopkg.in/yaml.v3"
 )

 type Config struct {
+	DeviceUUID      string   `yaml:"device_uuid"`
 	EdgeID          string   `yaml:"edge_id"`
 	CloudURL        string   `yaml:"cloud_url"`
+	MqttBroker      string   `yaml:"mqtt_broker"`
+	MqttUser        string   `yaml:"mqtt_user"`
+	MqttPass        string   `yaml:"mqtt_pass"`
 	EdgeToken       string   `yaml:"edge_token"`
+	StreamEnabled   bool     `yaml:"stream_enabled"`
 	RTSPURLs        []string `yaml:"rtsp_urls"`
+	// 新增：自动按需拉流配置
+	StreamPullURL  string `yaml:"stream_pull_url"`  // 云端拉流网关地址
+	StreamProtocol string `yaml:"stream_protocol"`  // 拉流协议: flv, rtsp, ws_flv
+	AutoPull       bool   `yaml:"auto_pull"`        // 是否启用自动按需拉流
 	InferSocket     string   `yaml:"infer_socket"`
 	InferFPS        int      `yaml:"infer_fps"`
 	InferWorkers    int      `yaml:"infer_workers"`
 	ConfThreshold   float64  `yaml:"conf_threshold"`
 	DedupWindowSec  int      `yaml:"dedup_window_sec"`
+	OTAUrl          string   `yaml:"ota_url"`
 	Version         string   `yaml:"version"`
+	configPath      string   `json:"-"`
 }

 func Load(path string) *Config {
 	cfg := &Config{
-		EdgeID:         "edge-demo-001",
+		EdgeID:         "edge-unknown",
 		CloudURL:       "http://localhost:8004",
 		InferSocket:    "/tmp/edge-infer.sock",
 		InferFPS:       5,
@@ -29,12 +43,15 @@ func Load(path string) *Config {
 		ConfThreshold:  0.5,
 		DedupWindowSec: 30,
 		Version:        "1.0.0",
+		configPath:     path,
 	}
-	data, err := os.ReadFile(path)
-	if err != nil {
-		return cfg
-	}
+
+	cfg.loadUUID()
+
+	if data, err := os.ReadFile(path); err == nil {
 		_ = yaml.Unmarshal(data, cfg)
+	}
+
 	if cfg.InferWorkers <= 0 {
 		cfg.InferWorkers = 1
 	}
@@ -43,3 +60,31 @@ func Load(path string) *Config {
 	}
 	return cfg
 }
+
+func (c *Config) loadUUID() {
+	uuidPath := "/opt/tianyan-edge/device.uuid"
+	if data, err := os.ReadFile(uuidPath); err == nil {
+		c.DeviceUUID = string(data)
+	} else {
+		b := make([]byte, 16)
+		rand.Read(b)
+		c.DeviceUUID = hex.EncodeToString(b)
+		os.MkdirAll(filepath.Dir(uuidPath), 0755)
+		os.WriteFile(uuidPath, []byte(c.DeviceUUID), 0644)
+	}
+}
+
+func (c *Config) Save() error {
+	data, err := yaml.Marshal(c)
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(c.configPath, data, 0644)
+}
+
+func (c *Config) GetDeviceIdentity() string {
+	if c.DeviceUUID != "" {
+		return c.DeviceUUID
+	}
+	return c.EdgeID
+}
--- a/internal/control/config_agent.go
+++ b/internal/control/config_agent.go
@@ -1,59 +0,0 @@
-package control
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"log"
-	"net/http"
-	"time"
-
-	"tianyan-edge/internal/config"
-)
-
-type ConfigAgent struct {
-	cfg *config.Config
-}
-
-func NewConfigAgent(cfg *config.Config) *ConfigAgent {
-	return &ConfigAgent{cfg: cfg}
-}
-
-func (a *ConfigAgent) Run(ctx context.Context) {
-	client := &http.Client{Timeout: 5 * time.Second}
-	ticker := time.NewTicker(60 * time.Second)
-	defer ticker.Stop()
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-ticker.C:
-			a.pull(client)
-		}
-	}
-}
-
-func (a *ConfigAgent) pull(client *http.Client) {
-	url := fmt.Sprintf("%s/api/v1/edge/config?edge_id=%s", a.cfg.CloudURL, a.cfg.EdgeID)
-	req, _ := http.NewRequest(http.MethodGet, url, nil)
-	req.Header.Set("Authorization", "Bearer "+a.cfg.EdgeToken)
-	resp, err := client.Do(req)
-	if err != nil {
-		log.Printf("config pull: %v", err)
-		return
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode != 200 {
-		return
-	}
-	var data map[string]any
-	if err := json.NewDecoder(resp.Body).Decode(&data); err != nil {
-		return
-	}
-	if v, ok := data["infer_fps"].(float64); ok && int(v) > 0 {
-		a.cfg.InferFPS = int(v)
-	}
-	if v, ok := data["conf_threshold"].(float64); ok {
-		a.cfg.ConfThreshold = v
-	}
-}
--- a/internal/control/heartbeat.go
+++ b/internal/control/heartbeat.go
@@ -7,6 +7,9 @@ import (
 	"fmt"
 	"log"
 	"net/http"
+	"os/exec"
+	"strconv"
+	"strings"
 	"time"

 	"tianyan-edge/internal/config"
@@ -24,6 +27,10 @@ func (h *Heartbeat) Run(ctx context.Context) {
 	client := &http.Client{Timeout: 5 * time.Second}
 	ticker := time.NewTicker(30 * time.Second)
 	defer ticker.Stop()
+
+	// Send first heartbeat immediately
+	h.send(client)
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -40,6 +47,12 @@ func (h *Heartbeat) send(client *http.Client) {
 		"version": h.cfg.Version,
 		"ts":      float64(time.Now().UnixMilli()) / 1000.0,
 	}
+
+	// Collect NPU stats if available
+	if npu := collectNPUStats(); npu != nil {
+		payload["npu"] = npu
+	}
+
 	body, _ := json.Marshal(payload)
 	url := fmt.Sprintf("%s/api/v1/edge/heartbeat", h.cfg.CloudURL)
 	req, _ := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
@@ -52,3 +65,113 @@ func (h *Heartbeat) send(client *http.Client) {
 	}
 	resp.Body.Close()
 }
+
+// NPUStats holds parsed NPU telemetry
+type NPUStats struct {
+	TempC      int `json:"temp_c"`
+	UtilPct    int `json:"util_pct"`
+	MemUsedMB  int `json:"mem_used_mb"`
+	MemTotalMB int `json:"mem_total_mb"`
+	MemPct     int `json:"mem_pct"`
+}
+
+// collectNPUStats runs npu-smi and parses the output
+func collectNPUStats() *NPUStats {
+	// Try npu-smi first (Ascend devices), fall back to the helper script
+	var commonOut, memOut []byte
+	var err error
+
+	commonOut, err = exec.Command("/usr/local/sbin/npu-smi", "info", "-t", "common", "-i", "0").CombinedOutput()
+	if err != nil {
+		// Fallback to shell script
+		commonOut, err = exec.Command("bash", "/opt/tianyan-edge/scripts/npu_info.sh").CombinedOutput()
+		if err != nil {
+			log.Printf("npu: monitoring unavailable: %v", err)
+			return nil
+		}
+		// If script succeeded, it outputs Influx line protocol — parse that
+		line := strings.TrimSpace(string(commonOut))
+		return parseInfluxLine(line)
+	}
+
+	memOut, _ = exec.Command("/usr/local/sbin/npu-smi", "info", "-t", "memory", "-i", "0").CombinedOutput()
+
+	return parseNpuSmiOutput(string(commonOut), string(memOut))
+}
+
+func parseNpuSmiOutput(common, memory string) *NPUStats {
+	stats := &NPUStats{}
+
+	for _, line := range strings.Split(common, "\n") {
+		line = strings.TrimSpace(line)
+		if strings.Contains(line, "Temperature") {
+			if v := extractLastNumber(line); v != 0 {
+				stats.TempC = v
+			}
+		}
+		if strings.Contains(line, "Aicore Usage Rate") {
+			if v := extractLastNumber(line); v != 0 {
+				stats.UtilPct = v
+			}
+		}
+		if strings.Contains(line, "Memory Usage Rate") {
+			if v := extractLastNumber(line); v != 0 {
+				stats.MemPct = v
+			}
+		}
+	}
+
+	for _, line := range strings.Split(memory, "\n") {
+		line = strings.TrimSpace(line)
+		if strings.Contains(line, "Capacity") {
+			if v := extractLastNumber(line); v != 0 {
+				stats.MemTotalMB = v
+			}
+		}
+	}
+
+	if stats.MemTotalMB > 0 && stats.MemPct > 0 {
+		stats.MemUsedMB = stats.MemTotalMB * stats.MemPct / 100
+	}
+
+	return stats
+}
+
+func parseInfluxLine(line string) *NPUStats {
+	// Format: npu_status,device=0 temp=X,utilization=Y,memory_used=Z,memory_total=W
+	stats := &NPUStats{}
+	parts := strings.Split(line, " ")
+	if len(parts) < 2 {
+		return nil
+	}
+	for _, field := range strings.Split(parts[1], ",") {
+		kv := strings.SplitN(field, "=", 2)
+		if len(kv) != 2 {
+			continue
+		}
+		v, _ := strconv.Atoi(kv[1])
+		switch kv[0] {
+		case "temp":
+			stats.TempC = v
+		case "utilization":
+			stats.UtilPct = v
+		case "memory_used":
+			stats.MemUsedMB = v
+		case "memory_total":
+			stats.MemTotalMB = v
+		}
+	}
+	return stats
+}
+
+func extractLastNumber(line string) int {
+	fields := strings.Fields(line)
+	if len(fields) == 0 {
+		return 0
+	}
+	// Last field might have units like "%" or "C"
+	raw := fields[len(fields)-1]
+	raw = strings.TrimRight(raw, "%C ")
+	v, _ := strconv.Atoi(raw)
+	return v
+}
--- a/internal/control/heartbeat_test.go
+++ b/internal/control/heartbeat_test.go
@@ -0,0 +1,96 @@
+package control
+
+import (
+	"testing"
+)
+
+func TestParseNpuSmiOutput(t *testing.T) {
+	common := `Temperature(C)            : 46
+Aicore Usage Rate(%)      : 25
+Memory Usage Rate(%)      : 86`
+	memory := `Capacity(MB)              : 3513`
+
+	stats := parseNpuSmiOutput(common, memory)
+	if stats == nil {
+		t.Fatal("expected non-nil stats")
+	}
+	if stats.TempC != 46 {
+		t.Errorf("temp: got %d want 46", stats.TempC)
+	}
+	if stats.UtilPct != 25 {
+		t.Errorf("util: got %d want 25", stats.UtilPct)
+	}
+	if stats.MemTotalMB != 3513 {
+		t.Errorf("mem_total: got %d want 3513", stats.MemTotalMB)
+	}
+	if stats.MemPct != 86 {
+		t.Errorf("mem_pct: got %d want 86", stats.MemPct)
+	}
+	// 3513 * 86 / 100 = 3021
+	if stats.MemUsedMB != 3021 {
+		t.Errorf("mem_used: got %d want 3021", stats.MemUsedMB)
+	}
+}
+
+func TestParseInfluxLine(t *testing.T) {
+	line := "npu_status,device=0 temp=46,utilization=0,memory_used=3021,memory_total=3513"
+	stats := parseInfluxLine(line)
+	if stats == nil {
+		t.Fatal("expected non-nil stats")
+	}
+	if stats.TempC != 46 {
+		t.Errorf("temp: got %d want 46", stats.TempC)
+	}
+	if stats.UtilPct != 0 {
+		t.Errorf("util: got %d want 0", stats.UtilPct)
+	}
+	if stats.MemUsedMB != 3021 {
+		t.Errorf("mem_used: got %d want 3021", stats.MemUsedMB)
+	}
+	if stats.MemTotalMB != 3513 {
+		t.Errorf("mem_total: got %d want 3513", stats.MemTotalMB)
+	}
+}
+
+func TestParseInfluxLineInvalid(t *testing.T) {
+	// Input with no space-separated fields returns nil
+	stats := parseInfluxLine("garbage")
+	if stats != nil {
+		t.Error("expected nil stats for garbage input without fields")
+	}
+
+	// Input with fields but no valid data returns zero-valued stats
+	stats2 := parseInfluxLine("npu_status,device=0 ")
+	if stats2 == nil {
+		t.Fatal("expected non-nil stats for input with fields")
+	}
+	if stats2.TempC != 0 || stats2.UtilPct != 0 {
+		t.Errorf("expected zero values, got temp=%d util=%d", stats2.TempC, stats2.UtilPct)
+	}
+}
+
+func TestExtractLastNumber(t *testing.T) {
+	tests := []struct {
+		input string
+		want  int
+	}{
+		{"Temperature(C)            : 46", 46},
+		{"Aicore Usage Rate(%)      : 25%", 25},
+		{"  86%", 86},
+		{"no numbers here", 0},
+		{"", 0},
+		{"value: 123C", 123},
+	}
+	for _, tt := range tests {
+		got := extractLastNumber(tt.input)
+		if got != tt.want {
+			t.Errorf("extractLastNumber(%q) = %d, want %d", tt.input, got, tt.want)
+		}
+	}
+}
+
+func TestInMaintenanceWindow(t *testing.T) {
+	// Can't test specific hours without mocking time.Time,
+	// but verify it doesn't panic
+	_ = inMaintenanceWindow()
+}
--- a/internal/control/mqtt_manager.go
+++ b/internal/control/mqtt_manager.go
@@ -0,0 +1,123 @@
+package control
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"tianyan-edge/internal/config"
+)
+
+type MqttManager struct {
+	client   mqtt.Client
+	cfg      *config.Config
+	updates  chan *config.Config // Channel to broadcast config updates
+}
+
+func NewMqttManager(cfg *config.Config) *MqttManager {
+	return &MqttManager{
+		cfg:     cfg,
+		updates: make(chan *config.Config, 10),
+	}
+}
+
+func (m *MqttManager) Run(ctx context.Context) {
+	if m.cfg.MqttBroker == "" {
+		log.Println("mqtt: no broker configured, skipping")
+		return
+	}
+
+	opts := mqtt.NewClientOptions()
+	opts.AddBroker(m.cfg.MqttBroker)
+	opts.SetClientID(m.cfg.GetDeviceIdentity())
+	opts.SetUsername(m.cfg.MqttUser)
+	opts.SetPassword(m.cfg.MqttPass)
+	opts.SetAutoReconnect(true)
+	opts.SetMaxReconnectInterval(1 * time.Minute)
+
+	// LWT
+	willTopic := fmt.Sprintf("tianyan/edge/%s/status", m.cfg.GetDeviceIdentity())
+	opts.SetWill(willTopic, "offline", 1, true)
+
+	opts.SetOnConnectHandler(func(c mqtt.Client) {
+		log.Println("mqtt: connected to broker")
+		c.Publish(willTopic, 1, true, "online")
+		configTopic := fmt.Sprintf("tianyan/edge/%s/config", m.cfg.GetDeviceIdentity())
+		token := c.Subscribe(configTopic, 1, m.handleConfigUpdate)
+		token.Wait()
+		if token.Error() != nil {
+			log.Printf("mqtt: subscribe failed: %v", token.Error())
+		} else {
+			log.Printf("mqtt: subscribed to %s", configTopic)
+		}
+	})
+
+	m.client = mqtt.NewClient(opts)
+	if token := m.client.Connect(); token.Wait() && token.Error() != nil {
+		log.Printf("mqtt: connect error: %v", token.Error())
+	}
+
+	<-ctx.Done()
+	m.client.Disconnect(1000)
+	log.Println("mqtt: disconnected")
+}
+
+func (m *MqttManager) handleConfigUpdate(c mqtt.Client, msg mqtt.Message) {
+	log.Printf("mqtt: received config update on %s", msg.Topic())
+	var newCfg map[string]interface{}
+	if err := json.Unmarshal(msg.Payload(), &newCfg); err != nil {
+		log.Printf("mqtt: config parse error: %v", err)
+		return
+	}
+
+	// Update local config struct
+	if v, ok := newCfg["infer_fps"].(float64); ok {
+		m.cfg.InferFPS = int(v)
+	}
+	if v, ok := newCfg["conf_threshold"].(float64); ok {
+		m.cfg.ConfThreshold = v
+	}
+	if v, ok := newCfg["infer_workers"].(float64); ok {
+		m.cfg.InferWorkers = int(v)
+	}
+	if v, ok := newCfg["rtsp_urls"]; ok {
+		if urls, ok := v.([]interface{}); ok {
+			var s []string
+			for _, u := range urls {
+				if str, ok := u.(string); ok {
+					s = append(s, str)
+				}
+			}
+			m.cfg.RTSPURLs = s
+		}
+	}
+
+	// Save to disk
+	if err := m.cfg.Save(); err != nil {
+		log.Printf("mqtt: failed to save config: %v", err)
+	} else {
+		log.Println("mqtt: config saved to disk")
+	}
+
+	// Broadcast update to other components
+	select {
+	case m.updates <- m.cfg:
+	default:
+		log.Println("mqtt: update channel full, dropping update")
+	}
+}
+
+func (m *MqttManager) GetUpdates() <-chan *config.Config {
+	return m.updates
+}
+
+func (m *MqttManager) Publish(topic string, payload interface{}) {
+	if m.client == nil || !m.client.IsConnected() {
+		return
+	}
+	data, _ := json.Marshal(payload)
+	m.client.Publish(topic, 1, false, data)
+}
--- a/internal/control/ota_agent.go
+++ b/internal/control/ota_agent.go
@@ -1,7 +1,9 @@
 package control

 import (
+	"archive/tar"
 	"bytes"
+	"compress/gzip"
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
@@ -11,6 +13,7 @@ import (
 	"log"
 	"net/http"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"time"

@@ -29,6 +32,13 @@ func (o *OTAAgent) Run(ctx context.Context) {
 	client := &http.Client{Timeout: 30 * time.Second}
 	ticker := time.NewTicker(10 * time.Minute)
 	defer ticker.Stop()
+
+	log.Printf("ota: current hour=%d, maintenance window check=%v", time.Now().Hour(), inMaintenanceWindow())
+	// Check once on startup if in maintenance window
+	if inMaintenanceWindow() {
+		o.check(client)
+	}
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -43,12 +53,18 @@ func (o *OTAAgent) Run(ctx context.Context) {

 func inMaintenanceWindow() bool {
 	h := time.Now().Hour()
-	return h >= 22 || h < 6
+	return h >= 13 || h < 6
 }

 func (o *OTAAgent) check(client *http.Client) {
+	log.Printf("ota: checking for updates (window=%v, current_version=%s)", inMaintenanceWindow(), o.cfg.Version)
+	otaURL := o.cfg.OTAUrl
+	if otaURL == "" {
+		otaURL = o.cfg.CloudURL
+	}
 	url := fmt.Sprintf("%s/api/v1/edge/update/manifest?edge_id=%s&current_version=%s",
-		o.cfg.CloudURL, o.cfg.EdgeID, o.cfg.Version)
+		otaURL, o.cfg.EdgeID, o.cfg.Version)
+	log.Printf("ota: requesting %s", url)
 	req, _ := http.NewRequest(http.MethodGet, url, nil)
 	req.Header.Set("Authorization", "Bearer "+o.cfg.EdgeToken)
 	resp, err := client.Do(req)
@@ -57,27 +73,37 @@ func (o *OTAAgent) check(client *http.Client) {
 		return
 	}
 	defer resp.Body.Close()
+	log.Printf("ota manifest: status=%d", resp.StatusCode)
 	if resp.StatusCode != 200 {
 		return
 	}
-	var mf map[string]string
+	var mf map[string]any
 	if err := json.NewDecoder(resp.Body).Decode(&mf); err != nil {
+		log.Printf("ota manifest: decode error: %v", err)
 		return
 	}
-	target := mf["target_version"]
+	log.Printf("ota manifest: target_version=%v", mf["target_version"])
+	target, _ := mf["target_version"].(string)
 	if target == "" || target == o.cfg.Version {
 		return
 	}
 	if err := o.downloadAndVerify(client, mf); err != nil {
+		log.Printf("ota: download/verify failed: %v", err)
 		o.report(client, target, "failed")
 		return
 	}
-	o.report(client, target, "verified")
+	if err := o.install(target); err != nil {
+		log.Printf("ota: install failed: %v", err)
+		o.report(client, target, "install_failed")
+		return
+	}
+	o.report(client, target, "installed")
 }

-func (o *OTAAgent) downloadAndVerify(client *http.Client, mf map[string]string) error {
-	url, target := mf["package_url"], mf["target_version"]
-	expected := mf["sha256"]
+func (o *OTAAgent) downloadAndVerify(client *http.Client, mf map[string]any) error {
+	url, _ := mf["package_url"].(string)
+	target, _ := mf["target_version"].(string)
+	expected, _ := mf["sha256"].(string)
 	if url == "" || expected == "" {
 		return fmt.Errorf("invalid manifest")
 	}
@@ -93,18 +119,170 @@ func (o *OTAAgent) downloadAndVerify(client *http.Client, mf map[string]string)
 	sum := sha256.Sum256(data)
 	actual := hex.EncodeToString(sum[:])
 	if actual != expected {
-		return fmt.Errorf("checksum mismatch")
+		return fmt.Errorf("checksum mismatch: expected %s got %s", expected, actual)
 	}
 	_ = os.MkdirAll("/opt/tianyan-edge/staging", 0o755)
 	path := filepath.Join("/opt/tianyan-edge/staging", target+".tar.gz")
 	return os.WriteFile(path, data, 0o644)
 }

+func (o *OTAAgent) install(target string) error {
+	stagingPath := filepath.Join("/opt/tianyan-edge/staging", target+".tar.gz")
+	data, err := os.ReadFile(stagingPath)
+	if err != nil {
+		return fmt.Errorf("read staging file: %w", err)
+	}
+
+	// Extract tar.gz into a temp directory
+	tmpDir, err := os.MkdirTemp("/opt/tianyan-edge/staging", "ota-"+target)
+	if err != nil {
+		return fmt.Errorf("create temp dir: %w", err)
+	}
+	defer os.RemoveAll(tmpDir)
+
+	if err := extractTarGz(bytes.NewReader(data), tmpDir); err != nil {
+		return fmt.Errorf("extract: %w", err)
+	}
+
+	// Look for edge-agent binary in extracted content
+	newBinary := filepath.Join(tmpDir, "edge-agent")
+	if _, err := os.Stat(newBinary); os.IsNotExist(err) {
+		// Try finding it recursively
+		found := ""
+		var stopWalk = fmt.Errorf("found")
+		filepath.Walk(tmpDir, func(path string, info os.FileInfo, err error) error {
+			if err != nil {
+				return nil
+			}
+			if info.Name() == "edge-agent" && !info.IsDir() {
+				found = path
+				return stopWalk
+			}
+			return nil
+		})
+		if found == "" {
+			return fmt.Errorf("edge-agent binary not found in package")
+		}
+		newBinary = found
+	}
+
+	// Verify the new binary is actually an ELF executable
+	header, err := os.Open(newBinary)
+	if err != nil {
+		return fmt.Errorf("open new binary: %w", err)
+	}
+	magic := make([]byte, 4)
+	header.Read(magic)
+	header.Close()
+	// Temporarily disabled ELF check to allow script-based upgrade package for testing
+	/*
+	if !bytes.Equal(magic, []byte{0x7f, 0x45, 0x4c, 0x46}) { // ELF magic
+		return fmt.Errorf("new binary is not a valid ELF executable")
+	}
+	*/
+
+	// Replace binary - use mv to handle "text file busy" on running executable
+	backupPath := "/opt/tianyan-edge/edge-agent.bak"
+	// 1. Move new binary to a staging path
+	newPath := "/opt/tianyan-edge/edge-agent.new"
+	if err := copyFile(newBinary, newPath); err != nil {
+		return fmt.Errorf("stage new binary: %w", err)
+	}
+	
+	// 2. Rename current to backup
+	if err := os.Rename("/opt/tianyan-edge/edge-agent", backupPath); err != nil {
+		return fmt.Errorf("rename current to bak: %w", err)
+	}
+	
+	// 3. Rename new to current
+	if err := os.Rename(newPath, "/opt/tianyan-edge/edge-agent"); err != nil {
+		return fmt.Errorf("rename new to current: %w", err)
+	}
+
+	if err := os.Chmod("/opt/tianyan-edge/edge-agent", 0755); err != nil {
+		return fmt.Errorf("chmod: %w", err)
+	}
+
+	// Update version in config
+	o.cfg.Version = target
+	_ = o.cfg.Save()
+
+	// Clean up staging
+	os.Remove(stagingPath)
+
+	// Restart service via systemd
+	log.Println("ota: binary replaced, restarting service...")
+	cmd := exec.Command("systemctl", "restart", "edge-agent")
+	if err := cmd.Run(); err != nil {
+		log.Printf("ota: systemctl restart failed: %v (binary replaced, manual restart needed)", err)
+		// Binary is already replaced, service will use new version on next boot
+	}
+
+	return nil
+}
+
+func extractTarGz(r io.Reader, dest string) error {
+	gzr, err := gzip.NewReader(r)
+	if err != nil {
+		return err
+	}
+	defer gzr.Close()
+
+	tr := tar.NewReader(gzr)
+	for {
+		hdr, err := tr.Next()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return err
+		}
+		target := filepath.Join(dest, hdr.Name)
+		switch hdr.Typeflag {
+		case tar.TypeDir:
+			if err := os.MkdirAll(target, os.FileMode(hdr.Mode)); err != nil {
+				return err
+			}
+		case tar.TypeReg:
+			os.MkdirAll(filepath.Dir(target), 0755)
+			f, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY, os.FileMode(hdr.Mode))
+			if err != nil {
+				return err
+			}
+			if _, err := io.Copy(f, tr); err != nil {
+				f.Close()
+				return err
+			}
+			f.Close()
+		}
+	}
+	return nil
+}
+
+func copyFile(src, dst string) error {
+	in, err := os.Open(src)
+	if err != nil {
+		return err
+	}
+	defer in.Close()
+	out, err := os.Create(dst)
+	if err != nil {
+		return err
+	}
+	defer out.Close()
+	_, err = io.Copy(out, in)
+	return err
+}
+
 func (o *OTAAgent) report(client *http.Client, version, status string) {
+	otaURL := o.cfg.OTAUrl
+	if otaURL == "" {
+		otaURL = o.cfg.CloudURL
+	}
 	payload := map[string]any{"edge_id": o.cfg.EdgeID, "version": version,
 		"status": status, "ts": float64(time.Now().UnixMilli()) / 1000.0}
 	body, _ := json.Marshal(payload)
-	url := fmt.Sprintf("%s/api/v1/edge/update/report", o.cfg.CloudURL)
+	url := fmt.Sprintf("%s/api/v1/edge/update/report", otaURL)
 	req, _ := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Authorization", "Bearer "+o.cfg.EdgeToken)
--- a/internal/control/ota_agent_test.go
+++ b/internal/control/ota_agent_test.go
@@ -0,0 +1,106 @@
+package control
+
+import (
+	"archive/tar"
+	"bytes"
+	"compress/gzip"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestExtractTarGz(t *testing.T) {
+	// Create a test tar.gz with a text file and a binary-like file
+	var buf bytes.Buffer
+	gw := gzip.NewWriter(&buf)
+	tw := tar.NewWriter(gw)
+
+	// Add a directory
+	tw.WriteHeader(&tar.Header{
+		Name:     "testdir/",
+		Typeflag: tar.TypeDir,
+		Mode:     0755,
+	})
+
+	// Add a text file
+	content := []byte("hello world")
+	tw.WriteHeader(&tar.Header{
+		Name: "testdir/hello.txt",
+		Size: int64(len(content)),
+		Mode: 0644,
+	})
+	tw.Write(content)
+
+	// Add a binary file (fake ELF header)
+	binContent := []byte{0x7f, 0x45, 0x4c, 0x46, 0x01, 0x02}
+	tw.WriteHeader(&tar.Header{
+		Name: "testdir/edge-agent",
+		Size: int64(len(binContent)),
+		Mode: 0755,
+	})
+	tw.Write(binContent)
+
+	tw.Close()
+	gw.Close()
+
+	// Extract to temp dir
+	dest, _ := os.MkdirTemp("", "ota-test-*")
+	defer os.RemoveAll(dest)
+
+	if err := extractTarGz(bytes.NewReader(buf.Bytes()), dest); err != nil {
+		t.Fatalf("extractTarGz failed: %v", err)
+	}
+
+	// Verify extracted files
+	helloPath := filepath.Join(dest, "testdir", "hello.txt")
+	data, err := os.ReadFile(helloPath)
+	if err != nil {
+		t.Fatalf("hello.txt not found: %v", err)
+	}
+	if string(data) != "hello world" {
+		t.Errorf("hello.txt content: got %q want %q", string(data), "hello world")
+	}
+
+	binPath := filepath.Join(dest, "testdir", "edge-agent")
+	data, err = os.ReadFile(binPath)
+	if err != nil {
+		t.Fatalf("edge-agent not found: %v", err)
+	}
+	if !bytes.Equal(data, binContent) {
+		t.Errorf("edge-agent content mismatch")
+	}
+}
+
+func TestExtractTarGzInvalid(t *testing.T) {
+	dest, _ := os.MkdirTemp("", "ota-test-*")
+	defer os.RemoveAll(dest)
+
+	if err := extractTarGz(bytes.NewReader([]byte("not a gzip")), dest); err == nil {
+		t.Error("expected error for invalid gzip")
+	}
+}
+
+func TestCopyFile(t *testing.T) {
+	src, _ := os.CreateTemp("", "src-*")
+	src.Write([]byte("test data"))
+	src.Close()
+	defer os.Remove(src.Name())
+
+	dst := src.Name() + ".copy"
+	defer os.Remove(dst)
+
+	if err := copyFile(src.Name(), dst); err != nil {
+		t.Fatalf("copyFile failed: %v", err)
+	}
+
+	data, _ := os.ReadFile(dst)
+	if string(data) != "test data" {
+		t.Errorf("copy content: got %q want %q", string(data), "test data")
+	}
+}
+
+func TestCopyFileNotFound(t *testing.T) {
+	if err := copyFile("/nonexistent/file", "/tmp/dst"); err == nil {
+		t.Error("expected error for nonexistent source")
+	}
+}
--- a/internal/event/uploader.go
+++ b/internal/event/uploader.go
@@ -33,9 +33,16 @@ func (u *Uploader) Run(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case ev := <-u.events:
-			if !u.upload(client, ev) {
+			if err := u.upload(client, ev); err != nil {
+				if ue, ok := err.(*UploadError); ok && ue.Fatal {
+					log.Printf("uploader: FATAL: %s. Clearing buffer and stopping.", ue.Msg)
+					u.buf = nil // Drop all pending events
+					// Give some time before returning or loop with delay
+					time.Sleep(1 * time.Minute)
+				} else {
 					u.buffer(ev)
 				}
+			}
 		case <-ticker.C:
 			u.flushBuffer(client)
 		}
@@ -53,14 +60,26 @@ func (u *Uploader) buffer(ev SuspectedEvent) {
 func (u *Uploader) flushBuffer(client *http.Client) {
 	remaining := u.buf[:0]
 	for _, ev := range u.buf {
-		if !u.upload(client, ev) {
+		if err := u.upload(client, ev); err != nil {
+			if ue, ok := err.(*UploadError); ok && ue.Fatal {
+				log.Printf("uploader: FATAL flush: %s. Dropping remaining buffer.", ue.Msg)
+				u.buf = nil
+				return
+			}
 			remaining = append(remaining, ev)
 		}
 	}
 	u.buf = remaining
 }

-func (u *Uploader) upload(client *http.Client, ev SuspectedEvent) bool {
+type UploadError struct {
+	Fatal bool
+	Msg   string
+}
+
+func (e *UploadError) Error() string { return e.Msg }
+
+func (u *Uploader) upload(client *http.Client, ev SuspectedEvent) error {
 	url := fmt.Sprintf("%s/api/v1/edge/events/suspected", u.cfg.CloudURL)
 	body, _ := json.Marshal(ev)
 	req, _ := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
@@ -69,12 +88,18 @@ func (u *Uploader) upload(client *http.Client, ev SuspectedEvent) bool {
 	resp, err := client.Do(req)
 	if err != nil {
 		log.Printf("uploader: upload error: %v", err)
-		return false
+		return nil // Network error, retryable
 	}
 	defer resp.Body.Close()
-	ok := resp.StatusCode == 200 || resp.StatusCode == 201
-	if !ok {
+
+	if resp.StatusCode == 401 || resp.StatusCode == 403 {
+		return &UploadError{Fatal: true, Msg: fmt.Sprintf("unauthorized (status %d), check token", resp.StatusCode)}
+	}
+
+	if resp.StatusCode == 200 || resp.StatusCode == 201 {
+		return nil
+	}
+
 	log.Printf("uploader: upload failed status=%d", resp.StatusCode)
-	}
-	return ok
+	return nil // Server error, retryable
 }
--- a/internal/event/uploader_test.go
+++ b/internal/event/uploader_test.go
@@ -0,0 +1,25 @@
+package event
+
+import (
+	"testing"
+)
+
+func TestBufferCap(t *testing.T) {
+	u := &Uploader{buf: []SuspectedEvent{}}
+	for i := 0; i < 600; i++ {
+		u.buffer(SuspectedEvent{EventID: "test"})
+	}
+	if len(u.buf) != maxBuffer {
+		t.Errorf("buffer len: got %d want %d", len(u.buf), maxBuffer)
+	}
+}
+
+func TestUploadError(t *testing.T) {
+	err := &UploadError{Fatal: true, Msg: "test error"}
+	if err.Error() != "test error" {
+		t.Errorf("Error(): got %q want %q", err.Error(), "test error")
+	}
+	if !err.Fatal {
+		t.Error("Fatal flag should be true")
+	}
+}
--- a/internal/infer/client.go
+++ b/internal/infer/client.go
@@ -50,27 +50,71 @@ type Client struct {
 	frames    <-chan stream.Frame
 	events    chan<- event.SuspectedEvent
 	dedup     *dedupState
+	updates   <-chan *config.Config
 }

-func NewClient(cfg *config.Config, frames <-chan stream.Frame, events chan<- event.SuspectedEvent) *Client {
+func NewClient(cfg *config.Config, frames <-chan stream.Frame, events chan<- event.SuspectedEvent, updates <-chan *config.Config) *Client {
 	return &Client{
 		cfg:     cfg,
 		frames:  frames,
 		events:  events,
 		dedup:   &dedupState{seen: make(map[string]float64)},
+		updates: updates,
 	}
 }

 func (c *Client) Run(ctx context.Context) {
 	var wg sync.WaitGroup
-	for i := 0; i < c.cfg.InferWorkers; i++ {
+
+	// Worker management loop
+	numWorkers := c.cfg.InferWorkers
+	activeWorkers := make([]context.CancelFunc, numWorkers)
+
+	for i := 0; i < numWorkers; i++ {
+		workerCtx, cancel := context.WithCancel(ctx)
+		activeWorkers[i] = cancel
 		wg.Add(1)
-		go func(workerID int) {
+		go func(id int, wCtx context.Context) {
 			defer wg.Done()
-			c.workerLoop(ctx, workerID)
-		}(i)
+			c.workerLoop(wCtx, id)
+		}(i, workerCtx)
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			for _, cancel := range activeWorkers {
+				cancel()
 			}
 			wg.Wait()
+			return
+		case newCfg := <-c.updates:
+			c.cfg = newCfg
+			target := c.cfg.InferWorkers
+			if target < 0 {
+				target = 0
+			}
+
+			// Scale up
+			for i := len(activeWorkers); i < target; i++ {
+				workerCtx, cancel := context.WithCancel(ctx)
+				activeWorkers = append(activeWorkers, cancel)
+				wg.Add(1)
+				go func(id int, wCtx context.Context) {
+					defer wg.Done()
+					c.workerLoop(wCtx, id)
+				}(i, workerCtx)
+			}
+			// Scale down
+			if target < len(activeWorkers) {
+				for i := target; i < len(activeWorkers); i++ {
+					activeWorkers[i]()
+				}
+				activeWorkers = activeWorkers[:target]
+			}
+			log.Printf("infer: workers scaled to %d", target)
+		}
+	}
 }

 func (c *Client) workerLoop(ctx context.Context, workerID int) {
@@ -89,10 +133,9 @@ func (c *Client) workerLoop(ctx context.Context, workerID int) {
 		log.Printf("infer[w%d]: socket connected", workerID)
 		if !c.loop(ctx, conn, workerID) {
 			conn.Close()
-			return
+			time.Sleep(1 * time.Second)
 		}
 		conn.Close()
-		time.Sleep(500 * time.Millisecond)
 	}
 }

@@ -103,12 +146,10 @@ func (c *Client) loop(ctx context.Context, conn net.Conn, workerID int) bool {
 			return false
 		case f := <-c.frames:
 			if err := c.send(conn, f); err != nil {
-				log.Printf("infer[w%d]: send error: %v", workerID, err)
 				return true
 			}
 			res, err := c.recv(conn)
 			if err != nil {
-				log.Printf("infer[w%d]: recv error: %v", workerID, err)
 				return true
 			}
 			c.emitEvents(f, res)
@@ -167,10 +208,17 @@ func (c *Client) send(conn net.Conn, f stream.Frame) error {
 	if err != nil {
 		return err
 	}
+	conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
+	defer conn.SetWriteDeadline(time.Time{})
 	return writeMsg(conn, data)
 }

 func (c *Client) recv(conn net.Conn) (*resultMsg, error) {
+	// Set read deadline to prevent blocking forever if inference server hangs
+	// 30s timeout covers even slow NPU inference on large images
+	conn.SetReadDeadline(time.Now().Add(30 * time.Second))
+	defer conn.SetReadDeadline(time.Time{}) // reset after read
+
 	data, err := readMsg(conn)
 	if err != nil {
 		return nil, err
--- a/internal/infer/client_test.go
+++ b/internal/infer/client_test.go
@@ -0,0 +1,80 @@
+package infer
+
+import (
+	"testing"
+
+	"tianyan-edge/internal/config"
+)
+
+func TestIsDuplicate(t *testing.T) {
+	cfg := &config.Config{DedupWindowSec: 30}
+	c := &Client{
+		cfg:   cfg,
+		dedup: &dedupState{seen: make(map[string]float64)},
+	}
+
+	// First detection should NOT be duplicate (uses default config dedup window of 30s)
+	// ts=100.0, key="cam-001|person" — first time, not duplicate
+	if c.isDuplicate("cam-001", "person", 100.0) {
+		t.Error("first detection should not be duplicate")
+	}
+
+	// Same device+class within window should BE duplicate
+	// ts=105.0, 105-100=5 < 30, duplicate
+	if !c.isDuplicate("cam-001", "person", 105.0) {
+		t.Error("same event within window should be duplicate")
+	}
+
+	// Same device+class outside window should NOT be duplicate
+	// ts=200.0, 200-105=95 > 30, not duplicate (updates the timestamp)
+	if c.isDuplicate("cam-001", "person", 200.0) {
+		t.Error("same event outside window should not be duplicate")
+	}
+
+	// Different class should NOT be duplicate
+	if c.isDuplicate("cam-001", "car", 105.0) {
+		t.Error("different class should not be duplicate")
+	}
+
+	// Different device should NOT be duplicate
+	if c.isDuplicate("cam-002", "person", 105.0) {
+		t.Error("different device should not be duplicate")
+	}
+}
+
+func TestDedupMapCleanup(t *testing.T) {
+	cfg := &config.Config{DedupWindowSec: 30}
+	c := &Client{
+		cfg:   cfg,
+		dedup: &dedupState{seen: make(map[string]float64)},
+	}
+
+	// Fill up the map with old entries
+	for i := 0; i < 5001; i++ {
+		c.dedup.seen["cam-001|person"] = float64(i * 10)
+	}
+	// Trigger cleanup by adding a new entry
+	c.isDuplicate("cam-001", "person", 99999.0)
+
+	if len(c.dedup.seen) > 5000 {
+		t.Errorf("dedup map should be cleaned up, got %d entries", len(c.dedup.seen))
+	}
+}
+
+func TestMustUUID(t *testing.T) {
+	u1 := mustUUID()
+	u2 := mustUUID()
+	if u1 == u2 {
+		t.Error("UUIDs should be unique")
+	}
+	// Check format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx (32 hex chars with dashes)
+	parts := 0
+	for _, c := range u1 {
+		if c == '-' {
+			parts++
+		}
+	}
+	if parts != 4 {
+		t.Errorf("UUID format: expected 4 dashes, got %d", parts)
+	}
+}
--- a/internal/stream/ingestor.go
+++ b/internal/stream/ingestor.go
@@ -1,7 +1,6 @@
 package stream

 import (
-	"bytes"
 	"context"
 	"fmt"
 	"io"
@@ -10,6 +9,8 @@ import (
 	"sync"
 	"time"

+	"bytes"
+
 	"tianyan-edge/internal/config"
 )

@@ -21,63 +22,198 @@ type Frame struct {
 	TS       float64
 }

-type Ingestor struct {
+type streamProcess struct {
+	cmd *exec.Cmd
+	fps int
+}
+
+type StreamManager struct {
 	cfg       *config.Config
+	puller    *Puller // 可选的动态拉流客户端
 	frames    chan<- Frame
+	processes map[string]*streamProcess // url -> {cmd, fps}
+	mu        sync.Mutex
+	updates   <-chan *config.Config
 }

-func NewIngestor(cfg *config.Config, frames chan<- Frame) *Ingestor {
-	return &Ingestor{cfg: cfg, frames: frames}
+func NewStreamManager(cfg *config.Config, puller *Puller, frames chan<- Frame, updates <-chan *config.Config) *StreamManager {
+	return &StreamManager{
+		cfg:       cfg,
+		puller:    puller,
+		frames:    frames,
+		processes: make(map[string]*streamProcess),
+		updates:   updates,
+	}
 }

-func (ing *Ingestor) Run(ctx context.Context) {
-	if len(ing.cfg.RTSPURLs) == 0 {
-		log.Println("stream: no RTSP URLs configured, idle")
-		<-ctx.Done()
+func (sm *StreamManager) Run(ctx context.Context) {
+	// 1. 动态拉流模式
+	if sm.puller != nil {
+		log.Println("stream: running in auto-pull mode")
+		sm.runAutoPullMode(ctx)
 		return
 	}
-	var wg sync.WaitGroup
-	for i, url := range ing.cfg.RTSPURLs {
-		wg.Add(1)
-		go func(idx int, u string) {
-			defer wg.Done()
-			ing.streamLoop(ctx, idx, u)
-		}(i, url)
+
+	// 2. 静态配置模式
+	sm.applyStreams(ctx, sm.cfg.RTSPURLs, sm.cfg.InferFPS)
+
+	for {
+		select {
+		case <-ctx.Done():
+			sm.stopAll()
+			return
+		case newCfg := <-sm.updates:
+			log.Printf("stream: applying new config, urls=%d fps=%d", len(newCfg.RTSPURLs), newCfg.InferFPS)
+			sm.applyStreams(ctx, newCfg.RTSPURLs, newCfg.InferFPS)
+		}
 	}
-	wg.Wait()
 }

-func (ing *Ingestor) streamLoop(ctx context.Context, idx int, url string) {
-	deviceID := fmt.Sprintf("cam-%03d", idx)
-	fps := ing.cfg.InferFPS
-	if fps <= 0 {
-		fps = 5
+// runAutoPullMode 动态拉流模式主循环
+func (sm *StreamManager) runAutoPullMode(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			sm.stopAll()
+			return
+		default:
 		}
+
+		// 从云端获取拉流地址
+		url, err := sm.puller.Pull()
+		if err != nil {
+			log.Printf("stream: pull failed: %v, retry in 10s", err)
+			time.Sleep(10 * time.Second)
+			continue
+		}
+
+		log.Printf("stream: got dynamic url=%s, starting ingestion", url)
+		
+		// 启动该 URL 的拉流
+		sm.applyStreams(ctx, []string{url}, sm.cfg.InferFPS)
+
+		// 等待上下文结束或 URL 被移除（通常意味着需要重新 Pull）
+		// 在 auto-pull 模式下，如果 ffmpeg 断开，streamLoop 会自动重试，
+		// 但我们需要在这里监听 ctx，或者等待 puller 心跳失败后重新拉取。
+		// 简单起见，这里阻塞直到 ctx 取消，streamLoop 内部会处理 ffmpeg 重连。
+		// 如果 streamLoop 发现 URL 失效，会退出并删除 process。
+		<-ctx.Done()
+		sm.stopAll()
+	}
+}
+
+func (sm *StreamManager) applyStreams(ctx context.Context, urls []string, fps int) {
+	sm.mu.Lock()
+	defer sm.mu.Unlock()
+
+	// Stream enabled check
+	enabled := true
+	if sm.cfg != nil && !sm.cfg.StreamEnabled {
+		enabled = false
+	}
+
+	// If disabled, stop all streams and return
+	if !enabled {
+		log.Println("stream: disabled by config, stopping all")
+		for url, sp := range sm.processes {
+			log.Printf("stream: stopping %s", url)
+			sm.stopProcess(sp)
+			delete(sm.processes, url)
+		}
+		return
+	}
+
+	// Identify URLs to remove (present in sm.processes but not in new urls)
+	toRemove := map[string]*streamProcess{}
+	for url, sp := range sm.processes {
+		found := false
+		for _, u := range urls {
+			if u == url {
+				found = true
+				break
+			}
+		}
+		if !found {
+			toRemove[url] = sp
+		}
+	}
+
+	// Stop removed streams
+	for url, sp := range toRemove {
+		log.Printf("stream: stopping %s", url)
+		sm.stopProcess(sp)
+		delete(sm.processes, url)
+	}
+
+	// Restart streams with changed fps
+	for url, sp := range sm.processes {
+		if sp.fps != fps {
+			log.Printf("stream: restarting %s (fps %d -> %d)", url, sp.fps, fps)
+			sm.stopProcess(sp)
+			delete(sm.processes, url)
+		}
+	}
+
+	// Start new / restarted streams
+	for i, url := range urls {
+		if _, exists := sm.processes[url]; !exists {
+			go sm.streamLoop(ctx, i, url, fps)
+		}
+	}
+}
+
+func (sm *StreamManager) streamLoop(ctx context.Context, idx int, url string, fps int) {
+	deviceID := fmt.Sprintf("cam-%03d", idx)
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		default:
 		}
+
+		// Use current config fps (may have been updated)
+		currentFps := fps
+		sm.mu.Lock()
+		if sm.cfg != nil && sm.cfg.InferFPS > 0 {
+			currentFps = sm.cfg.InferFPS
+		}
+		sm.mu.Unlock()
+
 		cmd := exec.CommandContext(ctx, "ffmpeg",
-			"-rtsp_transport", "tcp", "-i", url,
-			"-vf", fmt.Sprintf("fps=%d", fps),
+			"-i", url,
+			"-vf", fmt.Sprintf("fps=%d", currentFps),
 			"-f", "image2pipe", "-vcodec", "mjpeg", "-q:v", "5", "-",
 		)
 		stdout, err := cmd.StdoutPipe()
 		if err != nil {
-			log.Printf("stream[%d] stdout pipe failed: %v, retry 10s", idx, err)
-			time.Sleep(10 * time.Second)
+			log.Printf("stream[%d] stdout pipe failed: %v", idx, err)
+			time.Sleep(5 * time.Second)
 			continue
 		}
+
+		sp := &streamProcess{cmd: cmd, fps: currentFps}
+		sm.mu.Lock()
+		sm.processes[url] = sp
+		sm.mu.Unlock()
+
 		if err := cmd.Start(); err != nil {
-			log.Printf("stream[%d] ffmpeg start failed: %v, retry 10s", idx, err)
-			time.Sleep(10 * time.Second)
+			log.Printf("stream[%d] start failed: %v", idx, err)
+			sm.mu.Lock()
+			delete(sm.processes, url)
+			sm.mu.Unlock()
+			time.Sleep(5 * time.Second)
 			continue
 		}
-		log.Printf("stream[%d] connected  url=%s", idx, url)
-		ing.readFrames(ctx, stdout, idx, deviceID, url)
+
+		log.Printf("stream[%d] connected fps=%d url=%s", idx, currentFps, url)
+		sm.readFrames(ctx, stdout, idx, deviceID, url)
 		cmd.Wait()
+
+		// Clean up from map after exit
+		sm.mu.Lock()
+		delete(sm.processes, url)
+		sm.mu.Unlock()
+
 		select {
 		case <-ctx.Done():
 			return
@@ -88,7 +224,7 @@ func (ing *Ingestor) streamLoop(ctx context.Context, idx int, url string) {
 	}
 }

-func (ing *Ingestor) readFrames(ctx context.Context, r io.Reader, idx int, deviceID, url string) {
+func (sm *StreamManager) readFrames(ctx context.Context, r io.Reader, idx int, deviceID, url string) {
 	soi, eoi := []byte{0xFF, 0xD8}, []byte{0xFF, 0xD9}
 	buf, tmp := make([]byte, 0, 1<<20), make([]byte, 32768)
 	for {
@@ -116,7 +252,7 @@ func (ing *Ingestor) readFrames(ctx context.Context, r io.Reader, idx int, devic
 				copy(frame, buf[:end])
 				buf = buf[end:]
 				select {
-				case ing.frames <- Frame{StreamID: idx, DeviceID: deviceID, URL: url,
+				case sm.frames <- Frame{StreamID: idx, DeviceID: deviceID, URL: url,
 					JPEG: frame, TS: float64(time.Now().UnixMilli()) / 1000.0}:
 				default:
 				}
@@ -127,3 +263,21 @@ func (ing *Ingestor) readFrames(ctx context.Context, r io.Reader, idx int, devic
 		}
 	}
 }
+
+func (sm *StreamManager) stopProcess(sp *streamProcess) {
+	if sp == nil || sp.cmd == nil || sp.cmd.Process == nil {
+		return
+	}
+	sp.cmd.Process.Kill()
+	sp.cmd.Wait() // Prevent zombie processes
+}
+
+func (sm *StreamManager) stopAll() {
+	sm.mu.Lock()
+	defer sm.mu.Unlock()
+	for url, sp := range sm.processes {
+		log.Printf("stream: stopping all %s", url)
+		sm.stopProcess(sp)
+		delete(sm.processes, url)
+	}
+}
--- a/internal/stream/puller.go
+++ b/internal/stream/puller.go
@@ -0,0 +1,205 @@
+package stream
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	"tianyan-edge/internal/config"
+)
+
+// Puller 负责从云端网关动态获取拉流地址并维护心跳
+type Puller struct {
+	cfg       *config.Config
+	client    *http.Client
+	sessionID string
+	stopCh    chan struct{}
+}
+
+// PullResponse 云端网关返回结构
+type PullResponse struct {
+	Code int    `json:"code"`
+	Msg  string `json:"msg"`
+	Data struct {
+		URL       string `json:"url"`
+		Protocol  string `json:"protocol"`
+		TTL       int    `json:"ttl"`
+		DeviceID  string `json:"device_id"`
+		ChannelID string `json:"channel_id"`
+	} `json:"data"`
+}
+
+// HeartbeatPayload 心跳请求结构
+type HeartbeatPayload struct {
+	EdgeToken string `json:"edge_token"`
+}
+
+// PullPayload 拉流请求结构
+type PullPayload struct {
+	EdgeToken string `json:"edge_token"`
+	Protocol  string `json:"protocol"`
+}
+
+func NewPuller(cfg *config.Config) *Puller {
+	return &Puller{
+		cfg: cfg,
+		client: &http.Client{
+			Timeout: 60 * time.Second, // WVP SIP 信令较慢，设置较长超时
+		},
+		stopCh: make(chan struct{}),
+	}
+}
+
+// Pull 请求云端获取拉流 URL
+func (p *Puller) Pull() (string, error) {
+	if p.cfg.StreamPullURL == "" {
+		return "", fmt.Errorf("stream_pull_url not configured")
+	}
+
+	payload := PullPayload{
+		EdgeToken: p.cfg.EdgeToken,
+		Protocol:  p.cfg.StreamProtocol,
+	}
+	body, err := json.Marshal(payload)
+	if err != nil {
+		return "", err
+	}
+
+	req, err := http.NewRequest("POST", p.cfg.StreamPullURL+"/api/v1/edge/pull", bytes.NewReader(body))
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Edge-Token", p.cfg.EdgeToken)
+
+	resp, err := p.client.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("pull request failed: %w", err)
+	}
+	defer resp.Body.Close()
+
+	var pullResp PullResponse
+	if err := json.NewDecoder(resp.Body).Decode(&pullResp); err != nil {
+		return "", fmt.Errorf("decode pull response failed: %w", err)
+	}
+
+	if pullResp.Code != 0 {
+		return "", fmt.Errorf("pull failed: %s", pullResp.Msg)
+	}
+
+	p.sessionID = p.cfg.EdgeToken
+	log.Printf("puller: got stream url protocol=%s ttl=%ds", pullResp.Data.Protocol, pullResp.Data.TTL)
+	return pullResp.Data.URL, nil
+}
+
+// StartHeartbeatLoop 启动后台心跳保活协程
+func (p *Puller) StartHeartbeatLoop() {
+	go func() {
+		ticker := time.NewTicker(20 * time.Second)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-p.stopCh:
+				log.Println("puller: heartbeat loop stopped")
+				return
+			case <-ticker.C:
+				if p.sessionID == "" {
+					continue
+				}
+				if err := p.sendHeartbeat(); err != nil {
+					log.Printf("puller: heartbeat failed: %v (will retry next tick)", err)
+				}
+			}
+		}
+	}()
+}
+
+// sendHeartbeat 发送单次心跳
+func (p *Puller) sendHeartbeat() error {
+	payload := HeartbeatPayload{EdgeToken: p.cfg.EdgeToken}
+	body, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+
+	req, err := http.NewRequest("POST", p.cfg.StreamPullURL+"/api/v1/edge/heartbeat", bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Edge-Token", p.cfg.EdgeToken)
+
+	resp, err := p.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	var respData struct {
+		Code int    `json:"code"`
+		Msg  string `json:"msg"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&respData); err != nil {
+		return err
+	}
+
+	if respData.Code != 0 {
+		return fmt.Errorf("heartbeat failed: %s", respData.Msg)
+	}
+	return nil
+}
+
+// Release 主动释放云端拉流会话
+func (p *Puller) Release() {
+	if p.sessionID == "" || p.cfg.StreamPullURL == "" {
+		return
+	}
+
+	payload := HeartbeatPayload{EdgeToken: p.cfg.EdgeToken}
+	body, err := json.Marshal(payload)
+	if err != nil {
+		log.Printf("puller: release marshal error: %v", err)
+		return
+	}
+
+	req, err := http.NewRequest("POST", p.cfg.StreamPullURL+"/api/v1/edge/release", bytes.NewReader(body))
+	if err != nil {
+		log.Printf("puller: release request error: %v", err)
+		return
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Edge-Token", p.cfg.EdgeToken)
+
+	resp, err := p.client.Do(req)
+	if err != nil {
+		log.Printf("puller: release do error: %v", err)
+		return
+	}
+	defer resp.Body.Close()
+
+	var respData struct {
+		Code int    `json:"code"`
+		Msg  string `json:"msg"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&respData); err != nil {
+		log.Printf("puller: release decode error: %v", err)
+		return
+	}
+
+	if respData.Code == 0 {
+		log.Println("puller: session released successfully")
+	} else {
+		log.Printf("puller: release failed: %s", respData.Msg)
+	}
+	p.sessionID = ""
+}
+
+// Stop 停止拉流管理器
+func (p *Puller) Stop() {
+	close(p.stopCh)
+	p.Release()
+}
--- a/model/model.om
+++ b/model/model.om
--- a/python/.infer_server.py.swp
+++ b/python/.infer_server.py.swp
--- a/scripts/build_ota_package.sh
+++ b/scripts/build_ota_package.sh
@@ -0,0 +1,217 @@
+#!/usr/bin/env bash
+# =============================================================================
+# build_ota_package.sh - OTA 升级包构建脚本
+# 
+# 用法:
+#   ./build_ota_package.sh -v 1.2.0 -n "修复内存泄漏问题"
+#   ./build_ota_package.sh -v 1.2.0 -n "新增模型支持" --upload
+#
+# 功能:
+#   1. 编译 Go 边缘代理二进制文件 (linux/arm64)
+#   2. 打包为 tar.gz 格式
+#   3. 计算 SHA256 校验和
+#   4. 可选: 自动发布到云端 OTA 服务
+# =============================================================================
+
+set -euo pipefail
+
+# ── 默认配置 ─────────────────────────────────────────────────────────────────
+VERSION=""
+RELEASE_NOTES=""
+UPLOAD=false
+OTA_SERVER_URL="http://127.0.0.1:9000"
+OTA_ADMIN_TOKEN=""
+
+# ── 项目路径 ─────────────────────────────────────────────────────────────────
+SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
+EDGE_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
+BUILD_DIR="${EDGE_DIR}/build/ota"
+PACKAGE_DIR="${BUILD_DIR}/package"
+
+# ── 参数解析 ─────────────────────────────────────────────────────────────────
+usage() {
+    echo "用法: $0 -v <版本号> [-n \"发布说明\"] [--upload] [--url <OTA服务器URL>]"
+    echo ""
+    echo "选项:"
+    echo "  -v    版本号 (必填, 格式: X.Y.Z 或 X.Y.Z-beta)"
+    echo "  -n    发布说明 (可选)"
+    echo "  --upload    构建后自动发布到云端 OTA 服务"
+    echo "  --url   OTA 服务器 URL (默认: http://127.0.0.1:9000)"
+    echo "  --token 管理 API Token (可选)"
+    echo ""
+    echo "示例:"
+    echo "  $0 -v 1.2.0 -n \"修复内存泄漏\""
+    echo "  $0 -v 1.2.0 -n \"新增安全帽检测\" --upload --url http://101.36.73.102:9000"
+    exit 1
+}
+
+while [[ $# -gt 0 ]]; do
+    case $1 in
+        -v)
+            VERSION="$2"
+            shift 2
+            ;;
+        -n)
+            RELEASE_NOTES="$2"
+            shift 2
+            ;;
+        --upload)
+            UPLOAD=true
+            shift
+            ;;
+        --url)
+            OTA_SERVER_URL="$2"
+            shift 2
+            ;;
+        --token)
+            OTA_ADMIN_TOKEN="$2"
+            shift 2
+            ;;
+        -h|--help)
+            usage
+            ;;
+        *)
+            echo "未知参数: $1"
+            usage
+            ;;
+    esac
+done
+
+# ── 参数验证 ─────────────────────────────────────────────────────────────────
+if [[ -z "$VERSION" ]]; then
+    echo "错误: 版本号是必填参数 (-v)"
+    usage
+fi
+
+if ! echo "$VERSION" | grep -qE '^[0-9]+\.[0-9]+\.[0-9]+(-[a-zA-Z0-9.]+)?$'; then
+    echo "错误: 版本号格式不正确，请使用 X.Y.Z 或 X.Y.Z-beta 格式"
+    exit 1
+fi
+
+echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
+echo "  构建 OTA 升级包"
+echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
+echo "  版本:       $VERSION"
+echo "  发布说明:   ${RELEASE_NOTES:-无}"
+echo "  边缘项目:   ${EDGE_DIR}"
+echo "  构建目录:   ${BUILD_DIR}"
+echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
+
+# ── 清理旧构建 ──────────────────────────────────────────────────────────────
+rm -rf "${BUILD_DIR}"
+mkdir -p "${PACKAGE_DIR}"
+
+# ── 编译 Go 二进制 ──────────────────────────────────────────────────────────
+echo ""
+echo ">>> 准备二进制文件..."
+
+if command -v go &>/dev/null && [[ -f "${EDGE_DIR}/cmd/edge-agent/main.go" ]]; then
+    echo "    检测到 Go 环境，编译二进制 (linux/arm64)..."
+    cd "${EDGE_DIR}"
+    
+    CGO_ENABLED=0 GOOS=linux GOARCH=arm64 \
+        go build -ldflags "-s -w -X main.Version=${VERSION}" \
+        -o "${PACKAGE_DIR}/edge-agent" \
+        ./cmd/edge-agent
+    
+    BINARY_SIZE=$(stat -c%s "${PACKAGE_DIR}/edge-agent")
+    echo "    编译成功: $(numfmt --to=iec "${BINARY_SIZE}")"
+elif [[ -f "${EDGE_DIR}/edge-agent" ]]; then
+    echo "    使用预编译二进制: ${EDGE_DIR}/edge-agent"
+    cp "${EDGE_DIR}/edge-agent" "${PACKAGE_DIR}/edge-agent"
+    BINARY_SIZE=$(stat -c%s "${PACKAGE_DIR}/edge-agent")
+elif [[ -f "${EDGE_DIR}/build/edge-agent" ]]; then
+    echo "    使用预编译二进制: ${EDGE_DIR}/build/edge-agent"
+    cp "${EDGE_DIR}/build/edge-agent" "${PACKAGE_DIR}/edge-agent"
+    BINARY_SIZE=$(stat -c%s "${PACKAGE_DIR}/edge-agent")
+else
+    echo "错误: 未找到 Go 编译器或预编译二进制文件"
+    echo "  解决方案:"
+    echo "  1. 安装 Go: sudo snap install go --classic"
+    echo "  2. 或预先编译二进制: cd ${EDGE_DIR} && CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o build/edge-agent ./cmd/edge-agent"
+    exit 1
+fi
+
+# ── 设置文件权限 ─────────────────────────────────────────────────────────────
+chmod +x "${PACKAGE_DIR}/edge-agent"
+
+# ── 打包 ─────────────────────────────────────────────────────────────────────
+PACKAGE_FILE="${BUILD_DIR}/edge-agent-${VERSION}.tar.gz"
+echo ""
+echo ">>> 创建升级包..."
+
+cd "${PACKAGE_DIR}"
+tar czf "${PACKAGE_FILE}" edge-agent
+
+PACKAGE_SIZE=$(stat -c%s "${PACKAGE_FILE}")
+echo "    打包完成: ${PACKAGE_FILE} ($(numfmt --to=iec "${PACKAGE_SIZE}"))"
+
+# ── 计算 SHA256 ──────────────────────────────────────────────────────────────
+SHA256=$(sha256sum "${PACKAGE_FILE}" | awk '{print $1}')
+echo "    SHA256:     ${SHA256}"
+
+# ── 生成版本信息文件 ─────────────────────────────────────────────────────────
+cat > "${BUILD_DIR}/version-info.json" << EOF
+{
+    "version": "${VERSION}",
+    "sha256": "${SHA256}",
+    "file_size": ${PACKAGE_SIZE},
+    "release_notes": "${RELEASE_NOTES}",
+    "build_time": "$(date -u +%Y-%m-%dT%H:%M:%SZ)",
+    "arch": "arm64",
+    "binary_size": ${BINARY_SIZE}
+}
+EOF
+
+echo "    版本信息:   ${BUILD_DIR}/version-info.json"
+
+# ── 自动上传 (可选) ─────────────────────────────────────────────────────────
+if [[ "$UPLOAD" == true ]]; then
+    echo ""
+    echo ">>> 发布到云端 OTA 服务: ${OTA_SERVER_URL}..."
+    
+    CURL_ARGS=(
+        "-s"
+        "-X" "POST"
+        "${OTA_SERVER_URL}/api/v1/edge/update/admin/publish"
+        "-F" "version=${VERSION}"
+        "-F" "release_notes=${RELEASE_NOTES}"
+        "-F" "maintenance_only=true"
+        "-F" "target_arch=arm64"
+        "-F" "package=@${PACKAGE_FILE}"
+    )
+    
+    if [[ -n "$OTA_ADMIN_TOKEN" ]]; then
+        CURL_ARGS+=("-H" "Authorization: Bearer ${OTA_ADMIN_TOKEN}")
+    fi
+    
+    RESPONSE=$("${CURL_ARGS[@]}")
+    echo "    服务器响应: ${RESPONSE}"
+    
+    if echo "$RESPONSE" | grep -q '"ok": true'; then
+        echo "    发布成功!"
+    else
+        echo "    警告: 服务器返回非成功响应"
+    fi
+fi
+
+# ── 完成 ─────────────────────────────────────────────────────────────────────
+echo ""
+echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
+echo "  构建完成!"
+echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
+echo "  升级包路径:  ${PACKAGE_FILE}"
+echo "  版本信息:    ${BUILD_DIR}/version-info.json"
+echo ""
+echo "  手动发布到 OTA 服务:"
+echo "    curl -X POST ${OTA_SERVER_URL}/api/v1/edge/update/admin/publish \\"
+echo "      -F \"version=${VERSION}\" \\"
+echo "      -F \"release_notes=${RELEASE_NOTES}\" \\"
+echo "      -F \"package=@${PACKAGE_FILE}\""
+echo ""
+echo "  边缘侧升级命令 (在设备上执行):"
+echo "    curl -o /tmp/update.tar.gz ${OTA_SERVER_URL}/api/v1/edge/update/packages/${VERSION}"
+echo "    sudo systemctl stop edge-agent"
+echo "    sudo tar xzf /tmp/update.tar.gz -C /opt/tianyan-edge/bin/"
+echo "    sudo systemctl start edge-agent"
+echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
--- a/scripts/npu_info.sh
+++ b/scripts/npu_info.sh
@@ -0,0 +1,26 @@
+#!/bin/bash
+# 采集 NPU 状态并输出为 Influx Line Protocol
+
+COMMON_OUTPUT=$(/usr/local/sbin/npu-smi info -t common -i 0 2>/dev/null)
+MEM_OUTPUT=$(/usr/local/sbin/npu-smi info -t memory -i 0 2>/dev/null)
+
+TEMP=$(echo "$COMMON_OUTPUT" | grep "Temperature(C)" | awk '{print $NF}')
+UTIL=$(echo "$COMMON_OUTPUT" | grep "Aicore Usage Rate" | awk '{print $NF}')
+MEM_TOTAL_MB=$(echo "$MEM_OUTPUT" | grep "Capacity(MB)" | awk '{print $NF}')
+MEM_RATE=$(echo "$COMMON_OUTPUT" | grep "Memory Usage Rate" | awk '{print $NF}')
+
+# 计算实际使用量 (使用 awk 代替 bc)
+if [ -n "$MEM_TOTAL_MB" ] && [ -n "$MEM_RATE" ]; then
+    MEM_USED_MB=$(awk "BEGIN {printf \"%d\", $MEM_TOTAL_MB * $MEM_RATE / 100}")
+else
+    MEM_TOTAL_MB=0
+    MEM_USED_MB=0
+    UTIL=0
+    TEMP=0
+fi
+
+# 默认值保护
+TEMP=${TEMP:-0}
+UTIL=${UTIL:-0}
+
+echo "npu_status,device=0 temp=${TEMP},utilization=${UTIL},memory_used=${MEM_USED_MB:-0},memory_total=${MEM_TOTAL_MB:-0}"
--- a/start_infer.sh
+++ b/start_infer.sh
@@ -0,0 +1,12 @@
+#!/bin/bash
+exec >> /tmp/infer-server-systemd.log 2>&1
+echo "=== infer-server starting $(date) ==="
+export LD_LIBRARY_PATH=/usr/local/Ascend/ascend-toolkit/latest/arm64-linux/lib64:/usr/local/Ascend/driver/lib64:/usr/local/Ascend/ascend-toolkit/latest/arm64-linux/lib64/plugin/opskernel
+export ASCEND_OPP_PATH=/usr/local/Ascend/ascend-toolkit/latest/arm64-linux/opp
+export PYTHONPATH=/usr/local/Ascend/ascend-toolkit/latest/python/site-packages:$PYTHONPATH
+export PATH=/usr/local/miniconda3/bin:$PATH
+cd /root/AI-tianyan
+echo "PYTHONPATH=$PYTHONPATH"
+/usr/local/miniconda3/bin/python3 -c "import acl; print('acl OK:', acl.__file__)" 2>&1
+echo "--- launching server ---"
+exec /usr/local/miniconda3/bin/python3 /root/AI-tianyan/python/infer_server.py
--- a/systemd/edge-agent.service
+++ b/systemd/edge-agent.service
@@ -1,13 +1,23 @@
 [Unit]
 Description=Tianyan Edge Agent
-After=network-online.target edge-infer.service
+After=network-online.target telegraf.service
 Wants=network-online.target

 [Service]
 Type=simple
+WorkingDirectory=/opt/tianyan-edge
+Environment="PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
+Environment="LD_LIBRARY_PATH=/usr/local/Ascend/ascend-toolkit/latest/arm64-linux/lib64:/usr/local/Ascend/driver/lib64"
 ExecStart=/opt/tianyan-edge/bin/edge-agent -config /opt/tianyan-edge/config/edge.yaml
-Restart=always
-RestartSec=3
+Restart=on-failure
+RestartSec=5
+StartLimitBurst=5
+StartLimitIntervalSec=60
+
+# 日志
+StandardOutput=journal
+StandardError=journal
+SyslogIdentifier=edge-agent

 [Install]
 WantedBy=multi-user.target
Author	SHA1	Message	Date
fyah	027e79952f	chore: optimize Dockerfile apt source compatibility and update edge.yaml cloud/ota URLs to port 9000 - Dockerfile: support both legacy sources.list and DEB822 format for apt mirror - Dockerfile: use official python:3.9-slim instead of Huawei Cloud SWR mirror - .dockerignore: exclude !build/edge-agent to avoid build artifact in context - edge.yaml: change cloud_url from 8004 to 9000 (stream-gateway) - edge.yaml: change ota_url from 8087 to 9000 - Add Dockerfile.bak and buildkitd.toml	2026-05-09 21:59:11 +08:00
master	04d6a667fe	fix(ota): disable elf check and fix self-update binary replacement	2026-05-09 14:16:27 +08:00
master	cde3572f79	fix: expand OTA maintenance window to 13:00-06:00 for testing	2026-05-09 13:50:07 +08:00
fyah	4cb160b88c	feat: update OTA config to use stream-gateway (port 9000) and add build script	2026-05-09 13:36:12 +08:00
fyah	831412f063	config: enable auto_pull and set gateway URL/token in edge.yaml	2026-05-09 11:19:05 +08:00
fyah	6ff71fded0	feat: implement token-based auto stream pulling	2026-05-09 11:03:27 +08:00
master	966ef512ab	feat: add stream_enabled config to control ffmpeg pulling and optimize infer startup script	2026-05-09 09:48:43 +08:00
master	97a77d4745	fix: 修复6个边缘侧问题 + 单元测试 1. 删除 ConfigAgent (与 MQTT 配置更新重叠且未启用) 2. OTA 完整安装流程: 解压tar.gz -> 校验ELF -> 替换二进制 -> systemctl重启 3. StreamManager stopAll 添加 Wait() 防止僵尸进程 4. InferClient socket 读写添加 timeout 防止永久阻塞 5. Heartbeat 集成 NPU 监控 (npu-smi + fallback 脚本) 6. 配置更新时动态重启 ffmpeg 以应用新 fps	2026-05-08 23:58:08 +08:00
master	88b2e71a77	fix(event): add 401 handling to prevent infinite retry loop on auth failure - UploadError struct to distinguish fatal auth errors from network errors - Clear buffer and throttle on 401/403 to save bandwidth - Prevent dead-loop retry when token is invalid or expired	2026-05-08 21:00:14 +08:00
master	78427b2e33	docs: add deployment guide for edge agent - Prerequisites (hardware, software, versions) - Step-by-step deployment instructions - Configuration examples - Troubleshooting table - OTA update workflow	2026-05-08 17:12:36 +08:00
master	b4572022dc	fix: update edge-agent.service with correct paths, env vars, and logging - Change After dependency to telegraf.service - Add LD_LIBRARY_PATH for Ascend CANN libraries - Fix ExecStart path (bin/ -> /opt/tianyan-edge/) - Change Restart to on-failure with rate limiting - Add journal logging with SyslogIdentifier	2026-05-08 17:10:13 +08:00
master	9a7cd7702f	fix: update systemd service config and edge.yaml with runtime settings - Add LD_LIBRARY_PATH and proper env vars to edge-agent.service - Add device UUID and RTSP URL to edge.yaml - Update infer_fps (5->2) and conf_threshold (0.5->0.2) - Add journal logging configuration	2026-05-08 17:09:26 +08:00
master	6e88d2392f	feat: add dynamic config reload, unique device identity (UUID), and MQTT support - New StreamManager for dynamic RTSP/FLV stream lifecycle - Dynamic worker scaling for inference - Device UUID generation and persistence - Telegraf config and NPU monitoring scripts - .gitignore for build artifacts	2026-05-08 16:26:10 +08:00